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

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

RRPC指的是调用该接口向指定设备发送请求消息，并同步返回响应

在物联网场景下，如果想要做到Java服务与硬件同步通信的效果，那么一般会依赖MQTT来实现通信

比如Java服务向硬件发送请求，请求查询硬件相关信息

1. Java服务和硬件要提前订阅对应的Topic
2. Java服务先将消息发送到MQTT上（硬件订阅的Topic上）
3. 硬件订阅Topic收到消息后进行消费，消费完再发送ack响应消息到MQTT上（Java服务订阅的Topic上）

在这个同步通信的过程中，Java服务发送完消息是需要等待直到ack响应的，那么这个过程在Java服务端该如何实现这种等待/唤醒的模式呢？

本文就结合JUC组件来实现Java与硬件（通过MQTT）同步通信的组件

（为了简化流程，我们代码中使用阻塞队列代替MQTT）

整体流程

整体流程可以想象成远程调用的流程，只不过消费端是硬件，并且它们是通过MQTT转发消息来做到通信的

举例：把Java服务当作A端、把硬件当作B端，它们需要提前订阅MQTT上的topic

1. A端发送消息到B端订阅的Topic上，并进入等待状态（等待收到响应后唤醒）
2. B端订阅Topic收到消息后消费，响应并发送到A端订阅的Topic
3. A端订阅Topic的线程收到消息后进行解析，如果消息是当前节点需要处理的，则唤醒A端发送消息的线程

在这个过程中主要涉及四个线程：

1. A端发送消息的业务线程
2. B端接收消息并响应的线程
3. A端接收消息并唤醒的业务线程
4. A端定时删除超时的任务，防止内存泄漏

由于MQTT中间件太大，为了简化流程，我使用LinkedBlockingQueue进行模拟MQTT通信

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

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

流程代码：

1. 先启动B端消费线程
2. 在启动A端接收线程
3. 在模拟A端业务线程发送消息

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);
}

在sendMsg发送消息中，主要是创建DefaultFuture任务再发送消息，最后get阻塞等待结果到达时被唤醒

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;
}

DefaultFuture是我们要实现的任务组件、而MsgResponse是消息响应

类设计

Java服务发送完消息需要等待，直到响应到达；在这个过程中，非常与生产者、消费者模型类似

Java（生产者）发送完消息，进入等待状态，直到收到MQTT消息（相当于收到响应，消费完），唤醒接收响应

经典的生产者与消费者模型，可以考虑使用 synchronized + wait / notify 实现等待通知 或者 JUC下的Lock(AQS) + Condition 实现等待通知

但如果硬件一直不响应，请求一直等待会造成堆积，从而影响服务端性能，因此需要给等待设置超时时间，而synchronized并不支持，因此使用JUC下的 ReentrantLock（AQS阻塞队列）以及Condition（等待队列）

同时需要两个超时相关的字段：超时时间、时间单位

为了判断任务是否超时，需要记录任务开始的时间

在这个过程正好类似JUC下Future接口的流程，可以对Future接口进行实现

在分布式系统下，Java服务通常是多节点的，在同步通信的过程中由于MQTT是发布订阅模型，多节点都会收到响应，如何区分收到的响应消息是不是当前节点发送的呢？

因此需要增加消息的唯一标识（消息ID）用于判断消息，同时本地需要存储消息的唯一标识，存储可能是并发操作，因此考虑使用ConcurrentHashMap 相对线程安全的哈希表，能够根据消息ID快速判断当前节点是否需要接收

为了方便操作，我们将哈希表中的Value设置为我们的Future

由于实现Future接口，还可以取消任务，需要一个字段判断当前任务是否取消

同时需要一个字段存储收到的消息结果，并作为get的返回值

类设计完毕后，给我们的类取上名称——DefaultFuture

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();
}

方法实现

实现Future

Future接口提供5个方法：cancel取消任务、isCancelled任务是否被取消、isDone任务是否已完成、剩下两个get重载方法阻塞获取结果

我们依次进行实现：

boolean cancel(boolean mayInterruptIfRunning)

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

方法中的参数mayInterruptIfRunning为true时需要主动打断任务，因为我们通过MQTT通信，消息发送后无法打断，因此不需要管

只需要在容器中删除消息和标记消息已取消

boolean isCancelled()

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

返回是否取消的标记即可

boolean isDone();

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

是否完成只需要判断消息结果是否为空

V get() throws InterruptedException, ExecutionException;

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

默认超时时间为200ms

V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;

@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;
}

先进行参数校验、再判断是否完成，完成返回结果，未完成则加锁等待

由于只有收到消息才会调用唤醒方法并且也是只执行一次，这里没有使用循环防止虚假唤醒

其他方法

除了future接口的方法外，还需要其他方法来满足我们的需求

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

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

当收到消息时调用接收方法，received会先通过哈希表判断是否需要处理，如果删除获取的结果为空，说明当前节点没维护，不是当前节点不需要处理，否则需要去进行唤醒

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

唤醒前也需要加锁，并把结果赋值给字段msgResponse

补偿机制

在这个流程中，如果任务超时则会自动被唤醒，导致获取的结果为空，从而抛出异常

当超时的情况发生时，并没有清理哈希表中的记录，这就是常说的内存泄漏，当大量内存泄漏则会发生内存溢出

因此需要启动定时任务做补偿机制，循环判断任务是否已超时

	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();
    }

总结

本文结合JUC下的Lock、Condition与Future实现MQTT同步通信的组件

其中Lock与Condition是为了阻塞等待，但程序中的DefaultFuture是当作局部变量被使用的，并不存在并发

因此，如果只是为了等待而加锁是没必要的，可以考虑使用LockSupport.park/unpark进行等待

但是LockSupport.unpark唤醒时需要具体的线程，因此需要增加一个容器对任务与业务线程进行绑定存储

具体代码放在git仓库了，感兴趣的同学可以查看

🌠最后（点赞、收藏、关注求求啦~）

本篇文章被收入专栏 由点到线，由线到面，深入浅出构建Java并发编程知识体系，感兴趣的同学可以持续关注喔

本篇文章笔记以及案例被收入 Gitee-CaiCaiJava、 Github-CaiCaiJava，除此之外还有更多Java进阶相关知识，感兴趣的同学可以starred持续关注喔~

有什么问题可以在评论区交流，如果觉得菜菜写的不错，可以点赞、关注、收藏支持一下~

关注菜菜，分享更多技术干货，公众号：菜菜的后端私房菜