Android Netty集成

    implementation 'io.netty:netty-all:4.1.4.Final'
1

ExponentialBackOffRetry.java

package com.lanyu.cloud.netty;

import com.lanyu.cloud.utils.LogUtil;

import java.util.Random;

/**
 * <p>Retry policy that retries a set number of times with increasing sleep time between retries</p>
 */
public class ExponentialBackOffRetry implements RetryPolicy {

    private static final int MAX_RETRIES_LIMIT = 29;
    private static final int DEFAULT_MAX_SLEEP_MS = Integer.MAX_VALUE;

    private final Random random = new Random();
    private final long baseSleepTimeMs;
    private final int maxRetries;
    private final int maxSleepMs;

    public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries) {
        this(baseSleepTimeMs, maxRetries, DEFAULT_MAX_SLEEP_MS);
    }

    public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs) {
        this.maxRetries = maxRetries;
        this.baseSleepTimeMs = baseSleepTimeMs;
        this.maxSleepMs = maxSleepMs;
    }

    @Override
    public boolean allowRetry(int retryCount) {
        if (retryCount < maxRetries) {
            return true;
        }
        return false;
    }

    @Override
    public long getSleepTimeMs(int retryCount) {
        if (retryCount < 0) {
            throw new IllegalArgumentException("retries count must greater than 0.");
        }
        if (retryCount > MAX_RETRIES_LIMIT) {
            LogUtil.e(String.format("maxRetries too large (%d). Pinning to %d", maxRetries, MAX_RETRIES_LIMIT));
            retryCount = MAX_RETRIES_LIMIT;
        }
        long sleepMs = baseSleepTimeMs * Math.max(1, random.nextInt(1 << retryCount));
        if (sleepMs > maxSleepMs) {
            LogUtil.e(String.format("Sleep extension too large (%d). Pinning to %d", sleepMs, maxSleepMs));
            sleepMs = maxSleepMs;
        }
        return sleepMs;
    }
}
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

INettyMessageListener.java

package com.lanyu.cloud.netty;

public interface INettyMessageListener {
        void onReceive(String message);

        void onConnectSuccess();

        void onError();
    }

1
2
3
4
5
6
7
8
9
10

NettyChatClient.java

package com.lanyu.cloud.netty;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;

public class NettyChatClient {
    private static NettyChatClient mNettyChatClient;
    private static String mServerIp;
    private static int mPort;
    private Channel mChannel;
    /**
     * 重连策略
     */
    private RetryPolicy mRetryPolicy;
    private Bootstrap mBootstrap;

    public void init(INettyMessageListener messageListener) {
        mRetryPolicy = new ExponentialBackOffRetry(3000, Integer.MAX_VALUE, 3 * 1000);
        EventLoopGroup group = new NioEventLoopGroup();
        // bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.
        mBootstrap = new Bootstrap();
        mBootstrap.group(group)
                .option(ChannelOption.TCP_NODELAY, true)
                .channel(NioSocketChannel.class)
                .handler(new ClientHandlersInitializer());

    }

    public static NettyChatClient newInstance(String serverIp, int port) {
        if (mNettyChatClient == null) {
            mNettyChatClient = new NettyChatClient();
            mServerIp = serverIp;
            mPort = port;
        }
        return mNettyChatClient;
    }

    public static NettyChatClient getInstance() {
        return mNettyChatClient;
    }

    /**
     * 向远程TCP服务器请求连接
     */
    public void connect() {
        synchronized (mBootstrap) {
            ChannelFuture future = mBootstrap.connect(mServerIp, mPort);
            future.addListener(getConnectionListener());
            this.mChannel = future.channel();
        }
    }

    private ChannelFutureListener getConnectionListener() {
        return future -> {
            if (!future.isSuccess()) {
                future.channel().pipeline().fireChannelInactive();
            }
        };
    }

    private class ClientHandlersInitializer extends ChannelInitializer<SocketChannel> {


        @Override
        protected void initChannel(SocketChannel ch) {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast(new StringDecoder(StandardCharsets.UTF_8));
            pipeline.addLast(new StringEncoder(StandardCharsets.UTF_8));
            pipeline.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
            pipeline.addLast(new ReconnectHandler(mNettyChatClient));
        }

    }


    public RetryPolicy getRetryPolicy() {
        return mRetryPolicy;
    }

    public void sendMessage(String content) {
        if (mChannel != null) {
            mChannel.writeAndFlush(content);
        }
    }


}

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

ReconnectHandler.java

package com.lanyu.cloud.netty;

import android.annotation.SuppressLint;

import com.lanyu.cloud.utils.LogUtil;

import java.util.concurrent.TimeUnit;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoop;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;

@ChannelHandler.Sharable
public class ReconnectHandler extends SimpleChannelInboundHandler<String> {
    private int retries = 0;
    private RetryPolicy mRetryPolicy;
    private NettyChatClient mNettyChatClient;
    public ReconnectHandler(NettyChatClient nettyChatClient) {
        this.mNettyChatClient = nettyChatClient;
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        LogUtil.e("Successfully established a connection to the server.");
        retries = 0;
        ctx.fireChannelActive();
    }
    @SuppressLint("DefaultLocale")
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (retries == 0) {
            LogUtil.e("Lost the TCP connection with the server.");
            ctx.close();
        }
        boolean allowRetry = getRetryPolicy().allowRetry(retries);
        if (allowRetry) {
            long sleepTimeMs = getRetryPolicy().getSleepTimeMs(retries);
            LogUtil.e(String.format("Try to reconnect to the server after %dms. Retry count: %d.", sleepTimeMs, ++retries));
            final EventLoop eventLoop = ctx.channel().eventLoop();
            eventLoop.schedule(() -> {
                System.out.println("Reconnecting ...");
                mNettyChatClient.connect();
            }, sleepTimeMs, TimeUnit.MILLISECONDS);
        }
        ctx.fireChannelInactive();
    }

    /**
     * 心跳监测
     *
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            String eventType = null;
            switch (event.state()) {
                case READER_IDLE:
                    ctx.close();
                    eventType = "读空闲";
                    break;
                case WRITER_IDLE:
                    eventType = "写空闲";
                    break;
                case ALL_IDLE:
                    eventType = "读写空闲";
                    break;
                default:
            }
            LogUtil.e(ctx.channel().remoteAddress() + " " + eventType);
        }
    }


    /**
     * 收到消息后调用
     */

    @Override
    public void channelRead0(ChannelHandlerContext ctx, String message) {
        LogUtil.e("======================");
        LogUtil.e("收到消息" + message);
    }


    private RetryPolicy getRetryPolicy() {
        if (this.mRetryPolicy == null) {
            this.mRetryPolicy = mNettyChatClient.getRetryPolicy();
        }
        return this.mRetryPolicy;
    }
}
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

RetryPolicy.java

package com.lanyu.cloud.netty;

public interface RetryPolicy {

    /**
     * Called when an operation has failed for some reason. This method should return
     * true to make another attempt.
     *
     * @param retryCount the number of times retried so far (0 the first time)
     * @return true/false
     */
    boolean allowRetry(int retryCount);

    /**
     * get sleep time in ms of current retry count.
     *
     * @param retryCount current retry count
     * @return the time to sleep
     */
    long getSleepTimeMs(int retryCount);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

---

使用如下：

 private void initSocket() {
        NettyChatClient nettyChatClient = NettyChatClient.newInstance(Constant.SERVERIP, Constant.PORT);
        nettyChatClient.init(new INettyMessageListener() {
            @Override
            public void onReceive(String message) {
                for (INettyMessageListener nettyMessageListener : mIMessageListenerList) {
                    nettyMessageListener.onReceive(message);
                }
            }

            @Override
            public void onConnectSuccess() {
                for (INettyMessageListener nettyMessageListener : mIMessageListenerList) {
                    nettyMessageListener.onConnectSuccess();
                }
            }

            @Override
            public void onError() {
                for (INettyMessageListener nettyMessageListener : mIMessageListenerList) {
                    nettyMessageListener.onError();
                }
            }
        });
        nettyChatClient.connect();
    }
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

    NettyChatClient.getInstance().sendMessage(cmdJson.toJSONString());
1