JUC组件实战：实现RRPC(Java与硬件通过MQTT的同步通信)

    /**
     * 模拟MQTT通信时 A端 接收消息的 Topic
     */
    private static final LinkedBlockingQueue<String> mqttTopicA = new LinkedBlockingQueue<>();

    /**
     * 模拟MQTT通信时 B端 接收消息的 Topic
     */
    private static final LinkedBlockingQueue<String> mqttTopicB = new LinkedBlockingQueue<>();

public static void main(String[] args) {
    //1.开启消费线程 模拟B端消费消息
    Thread bConsumerThread = new Thread(() -> {
        while (true) {

            //获取消息
            String msgId = null;
            try {
                msgId = mqttTopicB.take();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            //消费..
            System.out.println("2." + Thread.currentThread().getName() + "消费消息：" + msgId);

            //消费完响应
            mqttTopicA.offer(msgId);

        }
    }, "B端消费线程");
    bConsumerThread.setDaemon(true);
    bConsumerThread.start();

    //2.开启接收线程 模拟A端接收消息
    Thread aReceivedThread = new Thread(() -> {
        while (true) {

            //获取消息
            String msgId = null;
            try {
                msgId = mqttTopicA.take();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            //唤醒业务处理线程
            MsgResponse msgResponse = new MsgResponse();
            msgResponse.setMsgId(msgId);
            //实际上消息通信会传输消息内容的，这里图方便 不想解析消息 就用msgID当作消息内容传输
            msgResponse.setMsgBodyJson("JSON消息内容");
            if (DefaultFuture.received(msgResponse)) {
                System.out.println("3." + Thread.currentThread().getName() + "唤醒消息：" + msgId);
            }
        }
    }, "A端接收消息线程");
    aReceivedThread.setDaemon(true);
    aReceivedThread.start();


    //3.业务处理线程发送消息
    String msgId = UUID.randomUUID().toString();
    System.out.println("1.A端业务线程开始发送消息");
    MsgResponse msgResponse = sendMsg(msgId);
    System.out.println("4.同步通信完毕，得到消息内容：" + msgResponse);
}

private static MsgResponse sendMsg(String msgId) {
    DefaultFuture future = new DefaultFuture(msgId);
    //模拟MQTT通信 发送消息给B端 让其消费消息
    mqttTopicB.offer(msgId);
    MsgResponse msgResponse;
    try {
        //阻塞等待 响应到达时被唤醒
        msgResponse = future.get();
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    } catch (ExecutionException e) {
        throw new RuntimeException(e);
    }
    return msgResponse;
}

public class DefaultFuture implements Future<MsgResponse> {

    public DefaultFuture(String msgId) {
        this(msgId, 200L, TimeUnit.MILLISECONDS);
    }

    public DefaultFuture(String msgId, long timeout, TimeUnit timeUnit) {
        this.msgId = msgId;
        this.timeout = timeout;
        this.timeUnit = timeUnit;

        futures.put(msgId, this);
        startTimeMillis = System.currentTimeMillis();
    }

    /**
     * 通过MQTT发布、订阅模型通信
     * 多Java节点的情况下无法分辨消息是不是当前节点发送的
     * 因此需要唯一标识 消息ID来判断
     * 并且本地还要进行存储，因此考虑使用KV的容器进行存储
     * 由于是会被并发访问，因此使用ConcurrentHashMap
     * K存储消息ID，V存储DefaultFuture，通过消息ID可以获取DefaultFuture
     */
    private static final Map<String, DefaultFuture> futures = new ConcurrentHashMap<>();

    /**
     * 消息唯一标识
     */
    private String msgId;

    /**
     * 用于判断任务是否被取消
     * true被取消  false未被取消
     */
    private boolean isCancel;

    /**
     * 启动时间用于判断是否超时
     */
    private long startTimeMillis;

    private long timeout;

    private TimeUnit timeUnit;

    /**
     * 消息通信的结果
     * 收到消息后 将消息封装成MsgResponse对象 存储到msgResponse中
     * 用于判断任务是否完成
     */
    private MsgResponse msgResponse;

    /**
     * 每个任务对应一把锁 和 一个等待队列
     * 用于实现等待通知模式
     */
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition();
}

@Override
public boolean cancel(boolean mayInterruptIfRunning) {
    //取消任务
    futures.remove(msgId);
    isCancel = true;
    return true;
}

@Override
public boolean isCancelled() {
    return isCancel;
}

@Override
public boolean isDone() {
    //结果不为空 说明消息通信完成
    return Objects.nonNull(msgResponse);
}

@Override
public MsgResponse get() throws InterruptedException, ExecutionException {
    return get(timeout, timeUnit);
}

@Override
public MsgResponse get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException {
    //参数校验
    if (timeout <= 0) {
        throw new RuntimeException("超时时间有误");
    }
    
    //覆盖超时字段
    this.timeout = timeout;
    this.timeUnit = unit;

    //如果任务完成就返回结果 否则阻塞等待任务完成
    if (isDone()) {
        return msgResponse;
    }

    //等待前要加锁
    lock.lock();
    try {
        condition.await(timeout, unit);
    } finally {
        lock.unlock();
    }

    return msgResponse;
}

public static boolean received(MsgResponse msg) {
    //接收时删除
    DefaultFuture future = futures.remove(msg.getMsgId());
    
    //当前节点不需要处理
    if (Objects.isNull(future)) {
        return false;
    }

    //唤醒
    future.doReceived(msg);
    return true;
}

private void doReceived(MsgResponse msg) {
    lock.lock();
    try {
        msgResponse = msg;
        condition.signal();
    } finally {
        lock.unlock();
    }
}

	public void startJob() {
        Thread timeoutJobThread = new Thread(() -> {
            while (true) {
                try {
                    for (DefaultFuture future : futures.values()) {
                        if (future == null || future.isDone()) {
                            continue;
                        }

                        //如果超时就去清理 以防内存泄漏
                        if (System.currentTimeMillis() - future.startTimeMillis > future.timeUnit.toMillis(timeout)) {
                            MsgResponse msgResponse = new MsgResponse();
                            msgResponse.setMsgId(future.msgId);
                            msgResponse.setMsgBodyJson("timeout");
                            DefaultFuture.received(msgResponse);
                        }
                    }
                    Thread.sleep(10000);
                } catch (Throwable e) {
                    //日志
                }
            }
        },"超时任务线程");
        
        timeoutJobThread.setDaemon(true);
        timeoutJobThread.start();
    }