public static void main(String[] args) {

        String topic = "defaultTopic";
        String content = "Message from 123xxx";
        int qos = 2;
        String broker = "tcp://127.0.0.1:1883";

        String clientId="123xxx";
        String userName = "admin";
        String pwd = "admin";
        try {
            /**
             * 不传递persistence默认使用文件保存未发送的消息，传递null使用内存保存未发送的消息
             */
            MqttClient mqttClient = new MqttClient(broker, clientId, null);
            MqttConnectOptions connOpts = new MqttConnectOptions();

            /**
             * 再次连接时是否清除上次的session，即重新连接时是否不接收未收到的消息。只有重新连接时才能够更改此参数
             */
            connOpts.setCleanSession(true);
            connOpts.setUserName(userName);
            connOpts.setPassword(pwd.toCharArray());
            // 发起连接
            mqttClient.connect(connOpts);

            // 订阅
            MqttCallback callback = new MessageProcess();
            mqttClient.setCallback(callback);
            mqttClient.subscribe(topic);
            // 发布
            // 1.创建需要发布的消息
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(qos);
            // 2.循环模拟发布
            while (true) {
                TimeUnit.SECONDS.sleep(5);
                mqttClient.publish(topic, message);
                System.out.println("Message published");
            }
        } catch (MqttException me) {
            System.out.println("reasonCode " + me.getReasonCode());
            System.out.println("message " + me.getMessage());
            System.out.println("localizedMessage " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("exception " + me.getMessage());
            me.printStackTrace();
        } catch (Exception e) {
            e.getMessage();
        }
    }

    static class MessageProcess implements MqttCallback {
        @Override
        public void connectionLost(Throwable throwable) {
            // 断线 手动重连
        }

        @Override
        public void messageArrived(String topic, MqttMessage message) {
            byte[] mess = message.getPayload();
            System.out.println("topic[" + topic + "]:" + new String(mess));
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
            // 确认信息已经传递完毕调用
        }
    }

public class NettyServer {
    private final int port;
 
    public NettyServer(int port) {
        this.port = port;
    }
 
    public void start() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
 
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            ServerBootstrap sb = new ServerBootstrap();
            sb.option(ChannelOption.SO_BACKLOG, 1024);
            sb.group(group, bossGroup) // 绑定线程池
                    .channel(NioServerSocketChannel.class) // 指定使用的channel
                    .localAddress(this.port)// 绑定监听端口
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 绑定客户端连接时候触发操作
 
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            System.out.println("收到新连接");
                            //websocket协议本身是基于http协议的，所以这边也要使用http解编码器
                            ch.pipeline().addLast(new HttpServerCodec());
                            //以块的方式来写的处理器
                            ch.pipeline().addLast(new ChunkedWriteHandler());
                            ch.pipeline().addLast(new HttpObjectAggregator(8192));
                            ch.pipeline().addLast(new MyWebSocketHandler());
                            ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", "WebSocket", true, 65536 * 10));
                        }
                    });
            ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
            System.out.println(NettyServer.class + " 启动正在监听： " + cf.channel().localAddress());
            cf.channel().closeFuture().sync(); // 关闭服务器通道
        } finally {
            group.shutdownGracefully().sync(); // 释放线程池资源
            bossGroup.shutdownGracefully().sync();
        }
    }
}

public class MyWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与客户端建立连接，通道开启！");

        //添加到channelGroup通道组
        MyChannelHandlerPool.channelGroup.add(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与客户端断开连接，通道关闭！");
        //添加到channelGroup 通道组
        MyChannelHandlerPool.channelGroup.remove(ctx.channel());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //首次连接是FullHttpRequest，处理参数 by zhengkai.blog.csdn.net
        if (null != msg && msg instanceof FullHttpRequest) {
            FullHttpRequest request = (FullHttpRequest) msg;
            String uri = request.uri();

            Map paramMap=getUrlParams(uri);
            System.out.println("接收到的参数是："+JSON.toJSONString(paramMap));
            //如果url包含参数，需要处理
            if(uri.contains("?")){
                String newUri=uri.substring(0,uri.indexOf("?"));
                System.out.println(newUri);
                request.setUri(newUri);
            }

        }else if(msg instanceof TextWebSocketFrame){
            //正常的TEXT消息类型
            TextWebSocketFrame frame=(TextWebSocketFrame)msg;
            System.out.println("客户端收到服务器数据：" +frame.text());
            sendAllMessage(frame.text());
        }
        super.channelRead(ctx, msg);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {

    }

    private void sendAllMessage(String message){
        //收到信息后，群发给所有channel
        MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
    }

    private static Map getUrlParams(String url){
        Map<String,String> map = new HashMap<>();
        url = url.replace("?",";");
        if (!url.contains(";")){
            return map;
        }
        if (url.split(";").length > 0){
            String[] arr = url.split(";")[1].split("&");
            for (String s : arr){
                String key = s.split("=")[0];
                String value = s.split("=")[1];
                map.put(key,value);
            }
            return  map;

        }else{
            return map;
        }
    }
}