2020-05-04 结合Netty和moquette模拟MQTT订阅客户端

模拟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

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 204,189评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,577评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 150,857评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,703评论 1赞 276
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,705评论 5赞 366
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,620评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,995评论 3赞 396
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,656评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,898评论 1赞 298
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,639评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,720评论 1赞 330
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,395评论 4赞 319
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,982评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,953评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,195评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 44,907评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,472评论 2赞 342