찾다

 >  Q&A  >  본문

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일 전834

모든 응답(4)나는 대답할 것이다

  • 阿神

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

    서버에 인코더가 없어 전혀 보낼 수 없다고 대략 짐작할 수 있습니다.
    게시된 코드를 보면 클라이언트에 메시지를 보내는 방법에는 두 가지가 있음을 알 수 있습니다.

    으아악

    으아악

    네티에서는 모든 항목과 종료가 ByteBuf입니다. 작성자는 서버에 해당 인코더가 있는지 확인하고 문자열 ByteBuf으로 변환해야 합니다.

    회신하다
    0
  • 巴扎黑

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

    GatewayService는 채널을 저장할 때 강제로 SocketChannel 유형으로 변환할 필요가 없습니다.
    채널을 가져오고 writeAndFlush를 수행하는 데에는 문제가 없습니다.
    문제를 추가로 해결하려면
    1. ctx.writeAndFlush(msg);의 Promise 결과를 모니터링하세요(예:

    ). 으아악

    2. 클라이언트와 서버의 인코더와 디코더가 동일한지 주의하세요

    회신하다
    0
  • 天蓬老师

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

    고객 관리에는 ChannelGroup을 사용하는 것이 좋습니다. 리더 서버가 데이터를 푸시할 수 없는 이유는 인코더가 없기 때문일 수 있습니다. ctx.writeAndFlush는 문자열 유형을 직접 쓸 수 없습니다.

    회신하다
    0
  • PHPz

    PHPz2017-04-18 09:05:28

    안녕하세요. 코드는 다음과 같이 수정할 수 있습니다.
    핸들러의 코드는 다음과 같이 수정됩니다.
    @Override

    으아악

    게이트웨이서비스:

    공개 클래스 GatewayService {

    으아악

    }

    클라이언트를 순회하는 코드는 다음과 같습니다.
    공용 클래스 TimeTask가 Runnable을 구현합니다.{

    으아악

    // obj.writeAndFlush("안녕하세요, 서버 테스트 헤더 핑입니다");

    으아악

    }

    그렇습니다. 개인적으로 테스트해본 결과 작동합니다

    회신하다
    0
  • 취소회신하다