拆分推送服务

Pu Zhibing

2025-04-18 9063f00db220dd93c94674d394ddbb38f83abd17

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
package com.sinata.push.util.echo;
 
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
 
import java.net.InetSocketAddress;
import java.util.HashMap;
 
public class DiscardServerHandler extends SimpleChannelInboundHandler<String>  {
 
    private NettyServerController nettyServerController = new NettyServerController();
    
    public static boolean isdebug = true;
 
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
        if(isdebug) {
//            System.err.println(insocket.getAddress() + ": 收到客户端数据.......");
        }
        try {
            // 调用service
            nettyServerController.JudgeOperation(ctx, msg);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }
 
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
        if(isdebug) {
//            System.err.println(insocket.getAddress() + ": 收到客户端数据.......");
        }
        try {
            // 调用service
            nettyServerController.JudgeOperation(ctx, msg);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }
 
//    @Override
//    protected void messageReceived(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
//
//    }
 
//    @Override
//    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
//
//    }
 
//    @Override
//    protected void messageReceived(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
//
//    }
 
    /** 在连接被建立并且准备进行通信时被调用 */
    public void channelActive(final ChannelHandlerContext ctx) throws Exception {
        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
        if(isdebug) {
//            System.err.println(insocket.getAddress() + ": Connect successful......");
        }
    }
 
    /** 心跳 */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state().equals(IdleState.READER_IDLE))
            {
                //
            } 
            else if (event.state().equals(IdleState.WRITER_IDLE)) 
            {
                //
            } 
            else if (event.state().equals(IdleState.ALL_IDLE))
            {
                String msg = NettyMsg.setMsg(Method.ok, new HashMap<String, Object>());
                if(ctx != null && ctx.channel().isActive()) {
                    ctx.writeAndFlush(Unpooled.copiedBuffer((msg).getBytes()));
//                    System.err.println(msg);
                }
            }
        }
        //super.userEventTriggered(ctx, evt);
    }
 
    /** 连接处于不活跃时调用（连接关闭） **/
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if(isdebug) {
            InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
//            System.err.println(insocket.getAddress() + ": Disconnect connection......");
        }
        NettyChannelMap.remove(ctx);
        System.err.println("清除通道" + ctx);
//        super.channelInactive(ctx);
    }
 
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
 
    /** 处理方法是当出现Throwable对象才会被调用 **/
    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause) throws Exception {
        ctx.close();
    }
 
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        if(isdebug) {
            InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
//            System.err.println("close......." + insocket.getAddress());
        }
        ctx.close(promise);
    }
 
    public void read(ChannelHandlerContext ctx) throws Exception {
        ctx.read();
    }
 
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        ctx.write(msg, promise);
    }
 
//    @Override
//    protected void channelRead(ChannelHandlerContext ctx, String msg) throws Exception {
//        // TODO Auto-generated method stub
//        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
//        if(isdebug) {
//            System.out.println(insocket.getAddress() + ": 收到客户端数据.......");
//        }
//        try {
//            // 调用service
//            nettyServerController.JudgeOperation(ctx, msg);
//        } catch (Exception e) {
//            e.printStackTrace();
//        } finally {
//            ReferenceCountUtil.release(msg);
//        }
//    }
    
}