2020-05-04 结合Netty和moquette模拟MQTT发布客户端

模拟MQTT发布客户端

上代码：
1.pubClient

package com.MyMQTTServerAndClient.publishClient;
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 pubClient {
    public static void main(String[] args) {
        pubClient.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 PubMQTTClientHandler());
                }
            });
            ChannelFuture ch = b.connect("127.0.0.1", 7777).sync();
            ch.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
            group.shutdownGracefully();
        }
    }
}

2.PubMQTTClientHandler

package com.MyMQTTServerAndClient.publishClient;
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;
/**
 * 暂时只支持Qos【0,1】
 * 备注：qos=0情况下服务端接收到数据后MessageID会被设置为null【应该是编解码的时候被设置】
 * 该类负责发布话题
 */
public class PubMQTTClientHandler extends SimpleChannelInboundHandler<AbstractMessage> {

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

    private String userName="admin";

    private String passWord="password";

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

    private static AtomicBoolean firstPingResp=new AtomicBoolean(true);//第一次ping通
    //Qos=0的待发布的话题
    LinkedBlockingDeque<PublishMessage> zeroQosPubQueue=new LinkedBlockingDeque<>();
    //Qos=1的待发布的话题
    LinkedBlockingDeque<PublishMessage> oneQosPubQueue=new LinkedBlockingDeque<>();
    //Qos=1的发布话题的ack
    LinkedBlockingDeque<PubAckMessage> oneQosPubAckQueue=new LinkedBlockingDeque<>();
    //已经发送的topic key为messageId value为该PublishMessage
    ConcurrentHashMap<Integer,PublishMessage> alreadySendMap=new ConcurrentHashMap<>();

    ExecutorService executorService;

    CountDownLatch countDownLatch=new CountDownLatch(1);

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

    private Channel channel;

    private Boolean running=new Boolean(true);

    public PubMQTTClientHandler() {
        executorService= Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        //发布Qos=0的话题
        java.util.concurrent.Future<?> future1 = executorService.submit(() -> {
            while(running){
                while (!zeroQosPubQueue.isEmpty()) {
                    PublishMessage publishMessage = zeroQosPubQueue.removeFirst();
                    channel.writeAndFlush(publishMessage);
                    logger.info("publishMessage {} publish success {}", publishMessage.getTopicName(),publishMessage.getMessageID());
                }
                try {
                    //防止执行过快
                    countDownLatch.await(1000, MILLISECONDS);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        //发布Qos=1的话题
        java.util.concurrent.Future<?> future2 = executorService.submit(() -> {
            while(running){
                while (!oneQosPubQueue.isEmpty()) {
                    PublishMessage publishMessage = oneQosPubQueue.removeFirst();
                    Integer messageID = publishMessage.getMessageID();
                    //还没发送过该topic消息,需要加入map
                    if (!alreadySendMap.containsKey(messageID)) {
                        alreadySendMap.put(messageID, publishMessage);
                    }
                    channel.writeAndFlush(publishMessage);
                    logger.info("publishMessage {} publish success {}", publishMessage.getTopicName(),publishMessage.getMessageID());
                    oneQosPubQueue.addLast(publishMessage);
                    try {
                        //防止发送过快
                        SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        //接受Qos=1的话题的ack
        java.util.concurrent.Future<?> future3 = executorService.submit(() -> {
            while(running){
                while (!oneQosPubAckQueue.isEmpty()) {
                    PubAckMessage pubAckMessage = oneQosPubAckQueue.removeFirst();
                    if (alreadySendMap.containsKey(pubAckMessage.getMessageID())) {
                        //topic收到ack，需要从发送队列移除该topic
                        oneQosPubQueue.remove(alreadySendMap.get(pubAckMessage.getMessageID()));
                        alreadySendMap.remove(pubAckMessage.getMessageID());
                    } else {
                        logger.error("receive error PubAckMessage {}",
                                pubAckMessage.getMessageID());
                    }
                }
                try {
                    //防止运行过快
                    SECONDS.sleep(1);
                } 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.PUBACK:
                    processPubAck(ctx,msg);
                    break;
                default:
                    logger.info("message type "+((AbstractMessage) msg).getMessageType());
                    break;
            }
        }
    }

    /**
     * 发布的topic Qos=1时会收到pubAck
     * @param ctx
     * @param msg
     */
    private void processPubAck(ChannelHandlerContext ctx, AbstractMessage msg) {
        PubAckMessage pubAckMessage = (PubAckMessage) msg;
        boolean result = oneQosPubAckQueue.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()){
//            IntStream.rangeClosed(5,200).forEach((i)->{
////                publishTopic("mqttClient/topicOne"+i,AbstractMessage.QOSType.LEAST_ONE);
////            });
            publishTopic("mqttClient/topicOne1",AbstractMessage.QOSType.LEAST_ONE);
            publishTopic("mqttClient/topicOne2",AbstractMessage.QOSType.LEAST_ONE);
            publishTopic("mqttClient/topicTwo2",AbstractMessage.QOSType.LEAST_ONE);
            publishTopic("mqttClient/topicTwo3",AbstractMessage.QOSType.MOST_ONE);
            for(int i=0;i<5;i++){
                publishTopic("mqttClient/topicTwo4",AbstractMessage.QOSType.MOST_ONE);
            }
            logger.debug("first receive ping response , start publish topic" );
            firstPingResp.set(false);
        }else{
            logger.debug("ping response");
        }
    }

    /**
     * 发布主题【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 = oneQosPubQueue.offer(publishMessage);
            if(offer){
                logger.info("publishMessage {} insert oneQosPubQueue 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/3f2d9ccfe310

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