需求场景:
智能家居网关(以下简称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 map = new ConcurrentHashMap<>();
public static void addGatewayChannel(String id, SocketChannel gateway_channel){
map.put(id, gateway_channel);
}
public static Map 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 map = GatewayService.getChannels();
Iterator 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如何主动向客户端发送消息?
你好,你的代码可以这样修改:
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(); } } }
}
这样就可以了。本人亲测,可以
推荐使用ChannelGroup管理客户端。领导服务器端推送不出去可能是因为没有编码器的原因,ctx.writeAndFlush是不能直接写串类型的。
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.注意客户端和服务端两边的编码器和解码器是否相同
可以大概猜测到是服务端没有编码器,因此根本发不出去
从贴出的代码里可以看到代码里往客户端发消息时有两种方式:
obj.writeAndFlush("hello, it is Server test header ping");
和
ctx.writeAndFlush(msg);//Object
在netty里,进出的都是ByteBuf
,楼主应确定服务端是否有对应的编码器,将字符串转化为ByteBuf
。