Netty简单实现客户端与服务端收发消息

Netty简单实现客户端与服务端收发消息

这个小案例主要是实现netty收发消息,分为客户端,及服务端,以及包含了相关状态处理,主要的代码会放在最后 gitHub 地址上,有需要可以看一下
首先来简单说明一下消息收发流程

1.客户端构建登录请求对象,服务端判断登录是否成功,并绑定成功状态
2.客户端判断登录成功,构建消息请求对象接收消息
3.服务端接收消息后进行处理,并构建消息返回对象,进行返回
4.客户端接收消息返回对象处理,流程结束

一 创建消息对象及状态新增
收发消息对象
消息请求对象 :

@Data
public class MessageRequestPacket extends Packet {
    /**
     * 消息体
     */
    private String message;

    @Override
    public Byte getCommand() {
        return MESSAGE_REQUEST;
    }
}

消息返回对象 :

@Data
public class MessageResponsePacket extends Packet {
    /**
     * 消息体
     */
    private String message;

    @Override
    public Byte getCommand() {
        return MESSAGE_RESPONSE;
    }
}

指令新增 :

public interface Command {
    Byte LOGIN_REQUEST = 1;

    Byte LOGIN_RESPONSE = 2;

    Byte MESSAGE_REQUEST = 3;

    Byte MESSAGE_RESPONSE = 4;
}

解码处理 :

public class PacketCodeC {

    private static final int MAGIC_NUMBER = 0X12345678;

    public static final PacketCodeC INSTANCE = new PacketCodeC();

    private final Map<Byte,Class<? extends Packet>> packetTypeMap;
    private final Map<Byte,Serializer> serializerMap;

    public PacketCodeC() {
        packetTypeMap = new HashMap<>();
        packetTypeMap.put(LOGIN_REQUEST, LoginRequestPacket.class);
        packetTypeMap.put(LOGIN_RESPONSE, LoginResponsePacket.class);
        packetTypeMap.put(MESSAGE_REQUEST, MessageRequestPacket.class);
        packetTypeMap.put(MESSAGE_RESPONSE, MessageResponsePacket.class);

        serializerMap = new HashMap<>();
        JSONSerializer serializer = new JSONSerializer();
        serializerMap.put(serializer.getSerializerAlgorithm(),serializer);
    }

    /**
     * 编码
     * @param packet
     * @return
     */
    public ByteBuf encode(ByteBuf byteBuf,Packet packet) {

        byte[] bytes = Serializer.DEFAULT.serialize(packet);

        //魔数
        byteBuf.writeInt(MAGIC_NUMBER);
        //版本号
        byteBuf.writeByte(packet.getVersion());
        //序列化算法
        byteBuf.writeByte(Serializer.DEFAULT.getSerializerAlgorithm());
        //指令
        byteBuf.writeByte(packet.getCommand());
        //数据长度
        byteBuf.writeInt(bytes.length);
        //数据
        byteBuf.writeBytes(bytes);

        return byteBuf;
    }

    public Packet decode(ByteBuf byteBuf) {
        //跳过魔数
        byteBuf.skipBytes(4);
        //跳过版本号
        byteBuf.skipBytes(1);
        //序列化算法标识
        byte serializerAlgorithm = byteBuf.readByte();
        //指令
        byte command = byteBuf.readByte();
        //数据包长度
        int length = byteBuf.readInt();
        //数据
        byte[] bytes = new byte[length];
        byteBuf.readBytes(bytes);

        Class<? extends Packet> requestType = getRequestType(command);
        Serializer serializer = getSerializer(serializerAlgorithm);

        if (requestType != null && serializer != null){
            return serializer.deserialize(requestType,bytes);
        }

        return null;
    }

    private Serializer getSerializer(byte serializerAlgorithm) {
        return serializerMap.get(serializerAlgorithm);
    }

    private Class<? extends Packet> getRequestType(byte command) {
        return packetTypeMap.get(command);
    }
}

二 判断是否登录
1.登录成功标志位定义

public interface Attributes {
    AttributeKey<Boolean> LOGIN = AttributeKey.newInstance("login");
}

ClientHandler 登录成功状态处理

if (packet instanceof LoginRequestPacket) {
            LoginRequestPacket loginResponsePacket = (LoginRequestPacket) packet;

            if (loginResponsePacket.getSuccess().equals(Boolean.TRUE)) {
                LoginUtil.markAsLogin(ctx.channel());
                System.out.println(new Date() + ": 客户端登录成功");
            }else {
                System.out.println(new Date() + ": 客户端登录失败,失败原因 : " +
                        loginResponsePacket.getReason());
            }
        }

上面代码主要在于 LoginUtil.markAsLogin(ctx.channel());这部分 状态的修改

public class LoginUtil {

    public static void markAsLogin(Channel channel) {
        channel.attr(Attributes.LOGIN).set(Boolean.TRUE);
    }

    public static boolean hasLogin(Channel channel) {
        Attribute<Boolean> loginAttr = channel.attr(Attributes.LOGIN);
        return loginAttr.get() != null;
    }
}

三 客户端输入消息并发送

public class LoginClient {
    public static void main(String[] args) {
        //
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workerGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel sc) throws Exception {
                        sc.pipeline().addLast(new ClientHandler());
                    }
                });

        connect(bootstrap,"127.0.0.1",8000,null);
    }

    private static void connect(Bootstrap bootstrap,String host,Integer port,Integer retry) {
        bootstrap.connect(host,port).addListener(future -> {
            System.out.println("future状态 :" + future.isSuccess());
            if (future.isSuccess()) {
                Channel channel = ((ChannelFuture) future).channel();
                startConsoleThread(channel);
            }
        });
    }

    private static void startConsoleThread(Channel channel) {
        new Thread(() -> {
            while (!Thread.interrupted()) {
                System.out.println("channel状态 :" + channel);
                if (LoginUtil.hasLogin(channel)) {
                    System.out.println("输入消息发送至服务端 : ");
                    Scanner sc = new Scanner(System.in);
                    String line = sc.nextLine();

                    MessageRequestPacket packet = new MessageRequestPacket();
                    packet.setMessage(line);

                    ByteBuf byteBuf = PacketCodeC.INSTANCE.encode(channel.alloc().ioBuffer(), packet);
                    channel.writeAndFlush(byteBuf);
                }
            }
        }).start();
    }
}

四 服务端,客户端消息处理
服务端消息处理 :

if (packet instanceof MessageRequestPacket){
            MessageRequestPacket messageRequestPacket = (MessageRequestPacket) packet;
            System.out.println(new Date() + " : 收到客户端消息 : " + messageRequestPacket.getMessage());

            MessageResponsePacket responsePacket = new MessageResponsePacket();

            responsePacket.setMessage("服务端回复 : [ " + messageRequestPacket.getMessage() + " ] ");
            ByteBuf responseByteBuf = PacketCodeC.INSTANCE.encode(ctx.channel().alloc().ioBuffer(), responsePacket);
            ctx.channel().writeAndFlush(responseByteBuf);
        }

客户端消息处理 :

if (packet instanceof MessageResponsePacket) {
            MessageResponsePacket messageResponsePacket = (MessageResponsePacket) packet;
            System.out.println(new Date() + " : 收到服务端的消息 :" + messageResponsePacket.getMessage());
        }

总的来说流程也较为简单,并没有太复杂的处理,有什么问题的话,断点做下跟踪基本都能解决了

源码地址 : https://github.com/xiatiandao/Netty-IMServer/tree/master