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

Question

需求场景：智能家居网关(以下简称gateway)，需要和netty服务器通讯(以下简称netty)，netty和gateway之间需要保持长连接(换句话说，netty和gateway之间都会主动给对方发送消息) 碰到的问题:netty作为服务器端如何...

阿神 · Answer

It can be roughly guessed that the server does not have an encoder, so it cannot be sent out at all.
From the posted code, you can see that there are two ways to send messages to the client:

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

and

ctx.writeAndFlush(msg);//Object

In netty, everything comes and goes ByteBuf，楼主应确定服务端是否有对应的编码器，将字符串转化为ByteBuf.

巴扎黑 · Answer

GatewayService does not need to be forced to convert to SocketChannel type when saving the channel.
There is no problem in getting the channel and doing writeAndFlush.
In order to further troubleshoot the problem, you can do this,
1. Monitor the Promise result of ctx.writeAndFlush(msg);, such as

ctx.writeAndFlush(msg).addListener(new ChannelFutureListener() {

    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        if (future.isSuccess()) {
            // do sth
        } else {
            // do sth
        }
    }
});

2. Pay attention to whether the encoder and decoder on the client and server are the same

天蓬老师 · Answer

It is recommended to use ChannelGroup to manage clients. The reason why the leader server cannot push data may be because there is no encoder. ctx.writeAndFlush cannot directly write string types.

PHPz · Answer

Hello, your code can be modified like this:
Handler’s code is modified as follows:
@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 map = new ConcurrentHashMap();
public static void addGatewayChannel(String id, ChannelHandlerContext gateway_ctx){
    map.put(id, gateway_ctx);
}
public static Map getChannels(){
    return map;
}

public static Object getGatewayChannel(String id){
    return map.get(id);
}

public static void removeGatewayChannel(String id){
    map.remove(id);
}

}

The code for traversing the client is as follows:
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 map = GatewayService.getChannels();
            Iterator 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();
        }
    }
}

}

That’s it. I personally tested it, it works