物联网 基于netty构建mqtt协议规范(遗嘱与保留消息)

简述

遗嘱消息（Last Will） 和 保留消息（Retained Message） 是两个重要的特性

  • 遗嘱消息

客户端异常断开（非正常发送 DISCONNECT）时，Broker 自动发布一条预定义的消息到指定主题，通知其他客户端

  • 保留消息

发布者可以标记消息为“保留”，Broker 为每个主题存储最新的保留消息。新订阅该主题的客户端会立即收到这条保留消息

源码(netty-sample-03-Last)

https://gitee.com/kcnf-iot/iot-sample/tree/master/netty/netty-sample-03

要点总结

特性触发条件存储位置推送时机
遗嘱消息客户端异常断开（Channel inactive 且未发送 DISCONNECT）内存中关联 Channel异常断开时立即发布
保留消息PUBLISH 命令中 retain=1ConcurrentHashMap<String, String>新订阅时立即推送（若存在）

server代码

package com.jysemel.iot;

import com.jysemel.iot.handler.BrokerHandler;
import io.netty.bootstrap.ServerBootstrap;
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.NioServerSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import java.nio.charset.StandardCharsets;

public class WillRetainBroker {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new LineBasedFrameDecoder(1024));
                            pipeline.addLast(new StringDecoder(StandardCharsets.UTF_8));
                            pipeline.addLast(new StringEncoder(StandardCharsets.UTF_8));
                            pipeline.addLast(new BrokerHandler());
                        }
                    });

            ChannelFuture future = bootstrap.bind(8888).sync();
            System.out.println("╔════════════════════════════════════╗");
            System.out.println("║   MQTT Broker 已启动               ║");
            System.out.println("╠════════════════════════════════════╣");
            System.out.println("║  端口: 8888                        ║");
            System.out.println("║  支持命令:                         ║");
            System.out.println("║    • CONNECT - 连接（设置遗嘱）    ║");
            System.out.println("║    • SUB     - 订阅主题            ║");
            System.out.println("║    • PUB     - 发布消息            ║");
            System.out.println("║    • DISCONNECT - 正常断开         ║");
            System.out.println("║  核心功能:                         ║");
            System.out.println("║    ✓ 遗嘱消息（异常断开触发）      ║");
            System.out.println("║    ✓ 保留消息（新订阅者立即可见）  ║");
            System.out.println("╚════════════════════════════════════╝");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

client代码

package com.jysemel.iot;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
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.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.nio.charset.StandardCharsets;

public class WillRetainClient {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(new LineBasedFrameDecoder(1024));
                            p.addLast(new StringDecoder(StandardCharsets.UTF_8));
                            p.addLast(new StringEncoder(StandardCharsets.UTF_8));
                        }
                    });

            System.out.println("╔════════════════════════════════════╗");
            System.out.println("║   MQTT 发布者客户端（演示遗嘱+保留） ║");
            System.out.println("╚════════════════════════════════════╝");
            System.out.println("🔌 正在连接 Broker...");

            Channel channel = bootstrap.connect("127.0.0.1", 8888).sync().channel();
            System.out.println("✅ 连接成功！\n");

            Thread.sleep(1000);

            System.out.println("━━━━━━━━━━━━ 步骤1: 设置遗嘱消息 ━━━━━━━━━━━━");
            System.out.println("📝 遗嘱主题: /device/status");
            System.out.println("📝 遗嘱内容: 设备离线（异常断开时触发）");
            channel.writeAndFlush("CONNECT /device/status 设备离线\n").sync();
            Thread.sleep(2000);

            System.out.println("\n━━━━━━━━━━━━ 步骤2: 发布普通消息 ━━━━━━━━━━━━");
            channel.writeAndFlush("PUB /device/status 设备在线\n").sync();
            System.out.println("📤 已发布: 设备在线");
            Thread.sleep(1500);

            System.out.println("\n━━━━━━━━━━━━ 步骤3: 发布保留消息 ━━━━━━━━━━━━");
            System.out.println("💡 特点: 新订阅者会立即收到此消息");
            channel.writeAndFlush("PUB /device/status 温度:25℃ RETAINED\n").sync();
            System.out.println("📤 已发布: 温度:25℃ [保留]");
            Thread.sleep(3000);

            System.out.println("\n━━━━━━━━━━━━ 测试说明 ━━━━━━━━━━━━");
            System.out.println("✅ 现在启动 SubscriberClient，会立即收到【保留消息】");
            System.out.println("⚠️  按 Ctrl+C 强制关闭本程序，将触发【遗嘱消息】");
            System.out.println("💡 遗嘱消息会发送给所有订阅了 /device/status 的客户端\n");

            channel.closeFuture().sync();

        } finally {
            group.shutdownGracefully();
        }
    }
}



package com.jysemel.iot;

import com.jysemel.iot.handler.ClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import java.nio.charset.StandardCharsets;

public class SubscriberClient {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(new LineBasedFrameDecoder(1024));
                            p.addLast(new StringDecoder(StandardCharsets.UTF_8));
                            p.addLast(new StringEncoder(StandardCharsets.UTF_8));
                            p.addLast(new ClientHandler());
                        }
                    });

            System.out.println("╔════════════════════════════════════╗");
            System.out.println("║   MQTT 订阅者客户端                 ║");
            System.out.println("╚════════════════════════════════════╝");
            System.out.println("🔌 正在连接 Broker...");

            Channel channel = bootstrap.connect("127.0.0.1", 8888).sync().channel();
            System.out.println("✅ 连接成功！\n");

            Thread.sleep(500);

            System.out.println("📋 订阅主题: /device/status");
            System.out.println("💡 如果该主题有保留消息，将立即收到\n");
            channel.writeAndFlush("SUB /device/status\n");

            System.out.println("⏳ 等待接收消息...\n");
            System.out.println("💡 提示：");
            System.out.println("   1. 先运行 WillRetainClient 发布保留消息");
            System.out.println("   2. 再运行本程序，会立即收到保留消息");
            System.out.println("   3. 强制关闭 WillRetainClient 可看到遗嘱消息\n");

            channel.closeFuture().sync();

        } finally {
            group.shutdownGracefully();
        }
    }
}

验证结果

img

img

img

最近更新:
Contributors: kcnf