模拟MQTT订阅客户端【Qos暂时只支持1】

1.subClient

package com.MyMQTTServerAndClient.subscribeClient;
/*
 * Description TODO
 * On 4/27/2020 5:41 PM
 */
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import org.eclipse.moquette.parser.netty.MQTTDecoder;
import org.eclipse.moquette.parser.netty.MQTTEncoder;
public class subClient {
    public static void main(String[] args) {
        subClient.startMqttClient();
    }
    public static void startMqttClient(){
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group);
            b.channel(NioSocketChannel.class);
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel channel) throws Exception {
                    ChannelPipeline pipeline = channel.pipeline();
                    pipeline.addLast("log", new LoggingHandler());
                    pipeline.addLast("decoder", new MQTTDecoder());//解码
                    pipeline.addLast("encoder", new MQTTEncoder());//编码
                    pipeline.addLast("handler", new SubMQTTClientHandler());
                }
            });
            ChannelFuture ch = b.connect("127.0.0.1", 7777).sync();
            ch.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
            group.shutdownGracefully();
        }
    }
}

2.SubMQTTClientHandler

package com.MyMQTTServerAndClient.subscribeClient;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.ScheduledFuture;
import org.eclipse.moquette.proto.messages.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.IntStream;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
/**
 *
 * 备注：SubscribeMessage的Qos固定为1【服务端回复SubAckMessage】，如果设置为其他，会被编码器过滤掉
 * 该类负责订阅话题
 */
public class SubMQTTClientHandler extends SimpleChannelInboundHandler<AbstractMessage> {

    private static Logger logger = LoggerFactory.getLogger(SubMQTTClientHandler.class);

    private String userName="admin";

    private String passWord="password";

    ScheduledFuture<?> scheduledFuture;//取消客户端心跳用

    private static AtomicBoolean firstPingResp=new AtomicBoolean(true);//第一次ping通
//    //Qos=0的待订阅的话题
//    LinkedBlockingDeque<SubscribeMessage> zeroQosSubQueue=new LinkedBlockingDeque<>();
    //Qos=1的待订阅的话题
    LinkedBlockingDeque<SubscribeMessage> oneQosSubQueue=new LinkedBlockingDeque<>();
    //Qos=1的订阅话题的ack
    LinkedBlockingDeque<SubAckMessage> oneQosSubAckQueue=new LinkedBlockingDeque<>();
    //已经发送订阅请求的map key为messageId value为该SubscribeMessage【用于处理Qos=1下，是否成功收到订阅成功ack】
    ConcurrentHashMap<Integer,SubscribeMessage> alreadySendMap=new ConcurrentHashMap<>();
    //已经成功订阅的话题【本地打印查看】
    LinkedBlockingDeque<SubscribeMessage> alreadySubQueue=new LinkedBlockingDeque<>();

    ExecutorService executorService;

    List<java.util.concurrent.Future> FutureList=new ArrayList<>();

    private Channel channel;

    private Boolean running=new Boolean(true);

    public SubMQTTClientHandler() {
        executorService= Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        //订阅Qos=1的话题
        java.util.concurrent.Future<?> future2 = executorService.submit(() -> {
            while(running){
                while (!oneQosSubQueue.isEmpty()) {
                    SubscribeMessage subscribeMessage = oneQosSubQueue.removeFirst();
                    Integer messageID = subscribeMessage.getMessageID();
                    //还没订阅过该topic消息,需要加入map
                    if (!alreadySendMap.containsKey(messageID)) {
                        alreadySendMap.put(messageID, subscribeMessage);
                        channel.writeAndFlush(subscribeMessage);
                        logger.info("subscribeMessage {} send to server success id {}, qos {}", subscribeMessage.subscriptions().get(0).getTopicFilter(),subscribeMessage.getMessageID(),subscribeMessage.getQos());
                    }
                    oneQosSubQueue.addLast(subscribeMessage);//此时还没成功收到服务端订阅成功的ack，不能移除该消息【在另一个线程中判断是否收到对应的ack来决定是否移除】
                    try {
                        //防止发送过快
                        SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        //接受Qos=1订阅话题的ack
        java.util.concurrent.Future<?> future3 = executorService.submit(() -> {
            while(running){
                while (!oneQosSubAckQueue.isEmpty()) {
                    SubAckMessage subAckMessage = oneQosSubAckQueue.removeFirst();
                    if (alreadySendMap.containsKey(subAckMessage.getMessageID())) {
                        //加入订阅成功队列
                        alreadySubQueue.add(alreadySendMap.get(subAckMessage.getMessageID()));
                        //收到订阅成功ack，需要从请求订阅队列移除该订阅消息
                        oneQosSubQueue.remove(alreadySendMap.get(subAckMessage.getMessageID()));
                        alreadySendMap.remove(subAckMessage.getMessageID());
                    } else {
                        logger.error("receive error SubAckMessage {}",
                                subAckMessage.getMessageID());
                    }
                }
                try {
                    //防止运行过快
                    MILLISECONDS.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        //打印成功订阅的话题
        executorService.submit(()-> {
            while (running) {
                if(!alreadySubQueue.isEmpty()){
                    logger.info("alreadySubQueue queue size {}",alreadySubQueue.size());
                    alreadySubQueue.forEach((p)->{
                        logger.info("already sub topic is "+p.subscriptions().get(0).getTopicFilter());
                    });
                }
                try {
                    //防止运行过快
                    SECONDS.sleep(20);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

//        FutureList.add(future1);
        FutureList.add(future2);
        FutureList.add(future3);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("error message {}",cause);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.info("client "+ctx.channel().remoteAddress()+" active");
        //连接
        ConnectMessage connectMessage = new ConnectMessage();
        connectMessage.setProcotolVersion((byte) 3);
        connectMessage.setQos(AbstractMessage.QOSType.LEAST_ONE);
        connectMessage.setCleanSession(true);
        connectMessage.setClientID(String.valueOf(UUID.randomUUID()));
        connectMessage.setKeepAlive(60);
        connectMessage.setWillFlag(false);
        connectMessage.setUserFlag(true);
        connectMessage.setUsername(this.userName);
        connectMessage.setPasswordFlag(true);
        connectMessage.setPassword(this.passWord);
        ctx.writeAndFlush(connectMessage,ctx.newPromise().addListener((future)->{
            if(future.cause()==null){
               logger.info("send connectMessage success");
            }else{
               logger.error("send connectMessage error "+future.cause());
            }
        }));

    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.error("channelInactive " +ctx.channel().remoteAddress());
        boolean cancel = scheduledFuture.cancel(true);
        logger.info("{} client canal heart beat {}",ctx.channel().localAddress(),cancel);
        ctx.close(ctx.newPromise().addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                logger.info("channel {} close ",ctx.channel().localAddress());
            }
        }));
        if(this.channel!=null){
            channel=null;
        }
        //
        running=false;
        //
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, AbstractMessage msg) throws Exception {
        if(msg instanceof AbstractMessage){
            byte messageType = msg.getMessageType();
            switch(messageType){
                case AbstractMessage.CONNACK:
                    processConnectAckResp(ctx,msg);
                    break;
                case AbstractMessage.PINGRESP:
                    processPingResp();
                    break;
                case AbstractMessage.PUBLISH:
                    processPublish(ctx,msg);
                    break;
                case AbstractMessage.SUBACK:
                    processSubAck(ctx,msg);
                    break;
                default:
                    logger.info("message type "+((AbstractMessage) msg).getMessageType());
                    break;
            }
        }
    }

    /**
     * 收到服务端发来的该客户端话题订阅成功ack
     * @param ctx
     * @param msg
     */
    private void processSubAck(ChannelHandlerContext ctx, AbstractMessage msg) {
        SubAckMessage subAckMessage = (SubAckMessage) msg;
        if(!oneQosSubAckQueue.contains(subAckMessage)){
            oneQosSubAckQueue.addLast(subAckMessage);
            logger.error("receive SubAckMessage from server id {}",subAckMessage.getMessageID());
        }else{
            logger.error("receive error SubAckMessage from server,id {}",subAckMessage.getMessageID());
        }
    }

    /**
     * 收到其他客户端发布的消息-由服务端发送来
     * @param ctx
     * @param msg
     */
    private void processPublish(ChannelHandlerContext ctx, AbstractMessage msg) {
        PublishMessage publishMessage = (PublishMessage) msg;
        logger.info("receive topic from other client topic {} payload [{}]",publishMessage.getTopicName(),new String(publishMessage.getPayload().array()));
    }

    /**
     * 发布的topic Qos=1时会收到pubAck
     * @param ctx
     * @param msg
     */
    private void processPubAck(ChannelHandlerContext ctx, AbstractMessage msg) {
//        PubAckMessage pubAckMessage = (PubAckMessage) msg;
//        boolean result = oneQosSubAckQueue.offerLast(pubAckMessage);
//        if(result==true){
//            logger.info("receive PubAckMessage {} from MQTTServer,and insert queue success",pubAckMessage.getMessageID());
//        }else{
//            logger.error("failed insert PubAckMessage {} into queue",pubAckMessage.getMessageID());
//        }
    }
    //第一次接收到服务端pingResp后才会发布主题或者订阅主题
    private void processPingResp() {
        if(firstPingResp.get()){
            logger.debug("first receive ping response , start subscribe topic" );
            /**
             *   qos=1只能是1，SubscribeMessage构造器设置
             */
            subscribeTopics("mqttClient/topicTwo2");
            subscribeTopics("mqttClient/topicTwo3");
            subscribeTopics("mqttClient/topicTwo4");
            firstPingResp.set(false);
        }else{
            logger.debug("ping response");
        }
    }
    /**
     * 订阅话题【目前Qos支持1】  这里我们默认只能订阅一个话题
     */
    private void subscribeTopics(String topic) {
        SubscribeMessage subscribeMessage = new SubscribeMessage();
        /**
         *   只能是qos=1，SubscribeMessage构造器设置
         */
        subscribeMessage.addSubscription(new SubscribeMessage.Couple((byte) AbstractMessage.QOSType.LEAST_ONE.ordinal(),topic));
        subscribeMessage.setMessageID(new Random().nextInt(10000));
        boolean offer = oneQosSubQueue.offer(subscribeMessage);
        if(offer){
            logger.info("subscribeMessage {} insert zeroQosSubQueue success is {}",subscribeMessage.subscriptions().get(0).getTopicFilter(),subscribeMessage.getMessageID());
        }
    }

//    /**
//     * 发布主题【Qos目前只支持0、1】
//     * @param topic
//     * @param qos
//     */
//    private void publishTopic(String topic,AbstractMessage.QOSType qos) {
//        //这个字段需要设置，否者无法发送消息
//        PublishMessage publishMessage = new PublishMessage();
//        publishMessage.setTopicName(topic);
//        publishMessage.setQos(qos);
//        int random = new Random().nextInt(10000);//<2^32-1
//        /**
//         *   qos=0情况下服务端接收到数据后MessageID会被设置为null【应该是编解码的时候被设置】
//         */
//        publishMessage.setMessageID(random);
//        publishMessage.setRetainFlag(false);
//        publishMessage.setPayload(ByteBuffer.wrap("I am payload".getBytes()));
//        if(qos== AbstractMessage.QOSType.MOST_ONE){
//            boolean offer = zeroQosPubQueue.offer(publishMessage);
//            if(offer){
//                logger.info("publishMessage {} insert zeroQosPubQueue success ",publishMessage.getTopicName());
//            }
//        }else if(qos== AbstractMessage.QOSType.LEAST_ONE){
//            boolean offer = oneQosSubQueue.offer(publishMessage);
//            if(offer){
//                logger.info("publishMessage {} insert oneQosSubQueue success ",publishMessage.getTopicName());
//            }
//        }
//    }

    private void processConnectAckResp(ChannelHandlerContext ctx, Object msg) {
        ConnAckMessage ackMessage = (ConnAckMessage) msg;
        if(ackMessage.getReturnCode()==(byte)0){
            channel=ctx.channel();
            logger.info("get connectAck success, start heart beat" );
            //存
//            ctx.channel().attr(MyConstants.pingKey).setIfAbsent("firstPing");
            // 定时心跳任务
            scheduledFuture = ctx.channel().eventLoop().scheduleAtFixedRate(() -> {
                PingReqMessage pingReqMessage = new PingReqMessage();
                pingReqMessage.setQos(AbstractMessage.QOSType.MOST_ONE);//心跳暂时MOST_ONE（0）
                ctx.writeAndFlush(pingReqMessage, ctx.newPromise().addListener((future) -> {
                    if (future.cause() == null) {
                        logger.info("send heartBeat success");
                    } else {
                        logger.error("send heartBeat error");
                    }
                }));
            }, 2, 55, SECONDS);
        }else if(ackMessage.getReturnCode()==(byte)4){
            logger.error("get connectAck failed, cannot heart beat,username or password" );//
            ctx.close();
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
       ctx.fireUserEventTriggered(evt);
    }
}

下一篇：发布客户端
https://www.jianshu.com/p/99db962e4ad3