club_delivery/src/main/java/com/ljsd/netty/NettyClient.java

package com.ljsd.netty;

import com.google.protobuf.GeneratedMessage;
import com.ljsd.netty.cocdex.Tea;
import com.ljsd.netty.impl.NettyTCPClientHandler;
import com.ljsd.netty.impl.NettyTCPClientInitializer;
import com.ljsd.timer.ServerManager;
import com.mongodb.DBObject;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.atomic.AtomicInteger;


public class NettyClient {
    private static final Logger LOGGER = LoggerFactory.getLogger(NettyClient.class);

    private static AtomicInteger requestId = new AtomicInteger(0);
    private static Map<Integer, NettyTCPClientHandler> handlerMap = new HashMap<>();

    private static List<EventLoopGroup> groups = new ArrayList<>();

    public static int getRequestId() {
        return requestId.addAndGet(1);
    }


    private static Channel getChannel(int serverId) {
        Channel channel = null;
        List<DBObject> serverInfoList = ServerManager.servers;
        DBObject server = null;
        for(DBObject data : serverInfoList) {
            if(Integer.parseInt(data.get("server_id").toString()) == serverId) {
                server = data;
                break;
            }
        }
        if(server == null) {
            return null;
        }
        String ip = server.get("ip").toString();
        int port = Integer.parseInt(server.get("port").toString());

        Bootstrap bootstrap = null;
        try {
            EventLoopGroup group = new NioEventLoopGroup(1);
            bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new NettyTCPClientInitializer())
                    .option(ChannelOption.TCP_NODELAY, true);

            channel = bootstrap.connect(ip, port).sync().channel();
            NettyTCPClientHandler handler = channel.pipeline().get(NettyTCPClientHandler.class);
            handlerMap.put(serverId, handler);
            groups.add(group);
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("获取handler失败");
        }
        return channel;
    }

    public static Result sendMessage(int routerId, int requestId, int serverId, int msgId,
                                         GeneratedMessage generatedMessage) throws InterruptedException {
        byte[] message = wrappedBytes(routerId, msgId, generatedMessage);
        return sendMessage(requestId, serverId, message);
    }

    public static byte[] wrappedBytes(int uid, int msgId, GeneratedMessage generatedMessage) {
        byte[] byMessage;
        if (generatedMessage == null) {
            byMessage = new byte[0];
        } else {
            byMessage = generatedMessage.toByteArray();
        }
        byte[] bytes = new byte[byMessage.length + PackageConstant.UID_FIELD_LEN
                + PackageConstant.ROUTERTYPE_LEN + PackageConstant.MSG_ID_FIELD_LEN
                + PackageConstant.INDEXT_FIELD_LEN + PackageConstant.INDICATION_INDEX_FIELD_LEN
                + PackageConstant.TOKEN_FIELD_LEN];
        int pos = 0;
        Tea.intToByte(bytes, pos, uid);
        pos += PackageConstant.UID_FIELD_LEN;
        bytes[pos] = 1;
        pos += PackageConstant.ROUTERTYPE_LEN;
        Tea.intToByte(bytes, pos, msgId);
        pos += PackageConstant.MSG_ID_FIELD_LEN;
        Tea.intToByte(bytes, pos, 0);
        pos += PackageConstant.INDEXT_FIELD_LEN;
        Tea.shortToByte(bytes, pos, 0);
        pos += PackageConstant.INDICATION_INDEX_FIELD_LEN;
        Tea.intToByte(bytes, pos, 0);
        pos += PackageConstant.TOKEN_FIELD_LEN;

        System.arraycopy(byMessage, 0, bytes, pos, byMessage.length);
        return bytes;
    }

    private static Result sendMessage(int requestId, int serverId, byte[] data) throws InterruptedException {
        NettyTCPClientHandler handler = handlerMap.get(serverId);
        if(handler == null || !handler.getChannel().isOpen()) {
            Channel channel = getChannel(serverId);
            if(channel == null) {
                return null;
            }
            handler = handlerMap.get(serverId);
        }
        SynchronousQueue<Result> queue = handler.sendMsg(requestId, data);
        Result result = queue.take();
        return result;
    }

    public static Map<Integer, NettyTCPClientHandler> getHandlerMap() {
        return handlerMap;
    }

    public static List<EventLoopGroup> getGroups() {
        return groups;
    }
}