JAVA《多线程多人上线通知案例》

package com.wangbiao.palyermanager;

import com.wangbiao.player.Player;

/**
 * TODO
 *
 * @author wangbiao
 * @Title TODO
 * @module TODO
 * @description 多人在线管理器
 * @date 2021/4/19 13:26
 */
public interface PlayerManager {
    /**
     * 增加一个玩家对象。
     */
    void addPlayer(Player player) throws InterruptedException;

    /**
     * 根据用户名获取玩家对象。
     */
    Player getPlayer(String username);

    /**
     * 向系统中的所有玩家广播一条消息。
     */
    void broadcast(String message) throws InterruptedException;

}

package com.wangbiao.player;

/**
 * TODO
 *
 * @author wangbiao
 * @Title TODO
 * @module TODO
 * @description TODO
 * @date 2021/4/19 13:26
 */

/*
题目：实现Player和PlayerManager接口的功能。

要求：
1、Player对象以username为索引，且Player对象创建之后，username不会变化。   》容器
2、PlayerManager中的所有功能是线程安全的，可并发执行。                  多线程
3、PlayerManager每隔一分钟会将isOffline() == true的Player对象删除。  timetask》可升级为定时器
4、编写针对PlayerManager功能的单元测试，确保PlayerManager的功能正确。     观察者模式/监听
*/

public interface Player {
    /**
     * 用户名。
     */
     String getUsername();

    /**
     * 向玩家发送消息。
     */
    void write(String message);

    /**
     * 玩家是否掉线。
     */
    boolean isOffline();

}

package com.wangbiao;

import com.wangbiao.palyermanager.PlayerManager;
import com.wangbiao.player.Player;

import java.util.Iterator;
import java.util.Map;
import java.util.Observable;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;


/**
 * TODO
 *
 * @author wangbiao
 * @Title TODO
 * @module TODO
 * @description 线程实现
 * @date 2021/4/19 14:45
 */
public class SunCallable implements Callable<String> {
private  static ReentrantLock reentrantLock=new ReentrantLock();
    private static final  Object object = new Object();
    private  static  volatile   ConcurrentHashMap<String, Player> hashMap;

    private ThreadPoolExecutor taskExecutor;
    private CountDownLatch latch;
    private Player player;

    public SunCallable(ThreadPoolExecutor taskExecutor, CountDownLatch latch, Player player,ConcurrentHashMap<String, Player> hashMap) {
        this.latch = latch;
        this.taskExecutor = taskExecutor;
        this.player = player;
        this.hashMap=hashMap;
    }

    public  String call() throws Exception {

            try {
                SunCallable.PlayerManagerInstance playerManagerInstance=new PlayerManagerInstance();
                Thread.sleep(1000);
                playerManagerInstance.addPlayer(this.player);
                Thread.sleep(1000);
                playerManagerInstance.broadcast("管理器通知：昵称为《" + this.player.getUsername() + "》上线了");
            } finally {
                latch.countDown();
            }

        return "ok";
    }

//静态内部类
  static   class PlayerManagerInstance extends Observable implements PlayerManager {
        public synchronized   void addPlayer(Player player) throws InterruptedException {
                hashMap.put(player.getUsername(), player);
                System.out.println("管理器通知：昵称为《" + player.getUsername() + "》上线了");
                setChanged();
                notifyObservers(player);



        }

        public Player getPlayer(String username) {
                return hashMap.get(username);

        }

        /**
         * 向系统中的所有玩家广播一条消息。
         */
        public  synchronized   void broadcast(String message) throws InterruptedException {
                for (Iterator<Map.Entry<String, Player>> iterator = hashMap.entrySet().iterator(); iterator.hasNext(); ) {

                    Map.Entry<String, Player> next = iterator.next();
                    Player player = next.getValue();
//                    if (getPlayer(player.getUsername()) == player) {
//                        continue;
//                    }
                    Thread.sleep(1000);
                    player.write(message);
                }

        }
    }
}

package com.wangbiao;

import com.wangbiao.player.Player;
import com.wangbiao.player.Player1;
import com.wangbiao.player.PlayerProxy;
import jdk.nashorn.internal.parser.JSONParser;
import netscape.javascript.JSObject;

import java.util.Timer;
import java.util.concurrent.*;

/**
 * TODO
 *
 * @author wangbiao
 * @Title TODO
 * @module TODO
 * @description TODO
 * @date 2021/4/19 13:46
 */
public class PayerTest {
    private  static  volatile ConcurrentHashMap<String, Player> hashMap = new ConcurrentHashMap();

    public static void main(String[] args) throws InterruptedException {

        ThreadPoolExecutor taskExecutor = new ThreadPoolExecutor(500,1000,3000, TimeUnit.MINUTES,   new LinkedBlockingQueue<Runnable>());
        CountDownLatch latch = new CountDownLatch(10);
        PlayerProxy playerProxy=new PlayerProxy();
        for (int i = 1; i <=10 ; i++) {
//动态代理类在多线程高并发场景下，出现只有一个线程实力有效的情况，所以一个玩家还是一个线程任务比较安全
//            Player player=playerProxy.getPlayer(new Player1("玩家"+i));  //巨坑卡了我一个下午
            Player1 player1 = new Player1("玩家" + i);
            taskExecutor.submit(new SunCallable(taskExecutor,latch,player1,hashMap));
        }
        MyTimeTask myTimeTask = new MyTimeTask(hashMap);
        Timer myTimer=new Timer();
        myTimer.schedule(myTimeTask,6000,6000);
        latch.await(1,TimeUnit.MINUTES);
        taskExecutor.shutdown();
    }
}