suchen

Heim  >  Fragen und Antworten  >  Hauptteil

java - netty如何实现向客户端主动发送消息

需求场景:
智能家居网关(以下简称gateway),需要和netty服务器通讯(以下简称netty),netty和gateway之间需要保持长连接(换句话说,netty和gateway之间都会主动给对方发送消息)

碰到的问题:
netty作为服务器端如何主动的向gateway发送消息,我尝试当每个gateway连接到netty(TCP/IP)时使用一个map把该channelSocket的id和该channelSocket绑定在一起

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        String uuid = ctx.channel().id().asLongText();
        GatewayService.addGatewayChannel(uuid, (SocketChannel)ctx.channel());
        System.out.println("a new connect come in: " + uuid);
    }

GatewayService其实就是一个ConcurrentHashMap

public class GatewayService {
    
    private static Map<String, SocketChannel> map = new ConcurrentHashMap<>();
    
    public static void addGatewayChannel(String id, SocketChannel gateway_channel){
        map.put(id, gateway_channel);
    }
    
    public static Map<String, SocketChannel> getChannels(){
        return map;
    }

    public static SocketChannel getGatewayChannel(String id){
        return map.get(id);
    }
    
    public static void removeGatewayChannel(String id){
        map.remove(id);
    }
}

我在服务器端尝试每间隔一段时间loop这个ConcurrentHashMap如果里面已经有绑定的channelSocket,就使用write方法向客户端发送消息

Runnable sendTask = new Runnable() {
            @Override
            public void run() {
                sendTaskLoop:
                    for(;;){
                        System.out.println("task is beginning...");
                        try{
                            Map<String, SocketChannel> map = GatewayService.getChannels();
                            Iterator<String> it = map.keySet().iterator();
                            while (it.hasNext()) {
                                String key = it.next();
                                SocketChannel obj = map.get(key);
                                System.out.println("channel id is: " + key);
                                System.out.println("channel: " + obj.isActive());
                                obj.writeAndFlush("hello, it is Server test header ping");
                            }
                        }catch(Exception e){break sendTaskLoop;}
                        try {
                            Thread.sleep(5000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
            }
        };
        new Thread(sendTask).start();

理论上客户端应该是可以接受到我发送的消息,但是我观察了一下源代码,发现writeAndFlush这个方法最终会被handler触发,于是我又在handler中覆写了write方法

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        System.out.println("write handler");
        ctx.writeAndFlush(msg);
    }

可是最终结果客户端并没有收到任何消息,请问netty如何主动向客户端发送消息?

ringa_leeringa_lee2768 Tage vor835

Antworte allen(4)Ich werde antworten

  • 阿神

    阿神2017-04-18 09:05:28

    可以大概猜测到是服务端没有编码器,因此根本发不出去
    从贴出的代码里可以看到代码里往客户端发消息时有两种方式:

    obj.writeAndFlush("hello, it is Server test header ping");

    ctx.writeAndFlush(msg);//Object

    在netty里,进出的都是ByteBuf,楼主应确定服务端是否有对应的编码器,将字符串转化为ByteBuf

    Antwort
    0
  • 巴扎黑

    巴扎黑2017-04-18 09:05:28

    GatewayService这里保存channel的时候不需要强制转换成SocketChannel类型,
    拿到channel做writeAndFlush是没有问题的。
    为了进一步排查问题,可以这么做,
    1.对ctx.writeAndFlush(msg);的Promise结果做监听,比如

    ctx.writeAndFlush(msg).addListener(new ChannelFutureListener() {
    
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (future.isSuccess()) {
                // do sth
            } else {
                // do sth
            }
        }
    });

    2.注意客户端和服务端两边的编码器和解码器是否相同

    Antwort
    0
  • 天蓬老师

    天蓬老师2017-04-18 09:05:28

    推荐使用ChannelGroup管理客户端。领导服务器端推送不出去可能是因为没有编码器的原因,ctx.writeAndFlush是不能直接写串类型的。

    Antwort
    0
  • PHPz

    PHPz2017-04-18 09:05:28

    你好,你的代码可以这样修改:
    Handler的代码修改如下:
    @Override

    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        String uuid = ctx.channel().id().asLongText();
        GatewayService.addGatewayChannel(uuid, ctx);
        logger.info("a new connect come in: " + uuid);
        logger.info("a new connect come in remote ip: " + ctx.channel().remoteAddress());
    }
    
    

    GatewayService:

    public class GatewayService {

    private static Map<String, Object> map = new ConcurrentHashMap();
    public static void addGatewayChannel(String id, ChannelHandlerContext gateway_ctx){
        map.put(id, gateway_ctx);
    }
    public static Map<String, Object> getChannels(){
        return map;
    }
    
    public static Object getGatewayChannel(String id){
        return map.get(id);
    }
    
    public static void removeGatewayChannel(String id){
        map.remove(id);
    }

    }

    遍历客户端的代码如下:
    public class TimeTask implements Runnable{

    private static final Log logger = LogFactory.getLog(TcpCMDHandler.class);
    public static final String CHARSET = "UTF-8";
    @Override
    public void run() {
        sendTaskLoop:
        for(;;){
            System.out.println("task is beginning...");
            try{
                Map<String, Object> map = GatewayService.getChannels();
                Iterator<String> it = map.keySet().iterator();
                while (it.hasNext()) {
                    String key = it.next();
                    ChannelHandlerContext ctx = (ChannelHandlerContext)map.get(key);
                    logger.info("channel id is: " + key);
                    logger.info("channel: " + ctx.channel().isActive());
    
                    String bak = "hello, every one!";
                    byte[] responB = bak.getBytes(CHARSET);
        //            ByteBuf response = Unpooled.buffer(responB.length);
    
                    final ByteBuf time = ctx.alloc().buffer(responB.length);
                    time.writeBytes(responB);
    
                    ctx.writeAndFlush(time); // (3)
    

    // obj.writeAndFlush("hello, it is Server test header ping");

                }
            }catch(Exception e){
                break
                sendTaskLoop;
            }
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    }

    这样就可以了。本人亲测,可以

    Antwort
    0
  • StornierenAntwort