proto文件

Example.proto

package example2.proto;

message BaseData {

required Header header = 1;

extensions 100 to 99999;

}

enum Header {

//裝備升級

Msg1001 = 1001;

//裝備穿戴

Msg1002 = 1002;

//添加好友

Msg1003 = 1003;

}

Friend.proto

package example2.proto;

import "example2/proto/Example.proto";

extend BaseData {

optional Receive1003 receive1003 = 10031;

optional Send1003 send1003 = 10032;

}

message Receive1003 { 

  required int32 friendId = 1;

}

message Send1003 { 

  required int32 friendId = 1;

  required int32 state = 2;

}

NettyServer服務(wù)器

package example2.server;

import io.netty.bootstrap.ServerBootstrap;

import io.netty.channel.ChannelFuture;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.ChannelOption;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioServerSocketChannel;

import io.netty.handler.codec.protobuf.ProtobufDecoder;

import io.netty.handler.codec.protobuf.ProtobufEncoder;

import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;

import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;

import java.util.logging.Level;

import java.util.logging.Logger;

import com.google.protobuf.ExtensionRegistry;

import example2.proto.Equip;

import example2.proto.Example;

import example2.proto.Friend;

import example2.server.handler.ProtoBufServerHandler;

public class NettyServer {

private static final int PORT = 1588;

private static Logger logger = Logger.getLogger(NettyServer.class.getName());

public void start(int port) throws Exception {

EventLoopGroup bossGroup = new NioEventLoopGroup();

EventLoopGroup workerGroup = new NioEventLoopGroup();

try {

ServerBootstrap b = new ServerBootstrap();

b.group(bossGroup, workerGroup);

b.channel(NioServerSocketChannel.class);

b.option(ChannelOption.TCP_NODELAY, true);

b.option(ChannelOption.SO_BACKLOG, 1024);

b.childHandler(new ChannelInitializer<SocketChannel>() {

@Override

protected void initChannel(SocketChannel ch) throws Exception {

//decoded

ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());

ExtensionRegistry registry = ExtensionRegistry.newInstance();

Equip.registerAllExtensions(registry);

Friend.registerAllExtensions(registry);

ch.pipeline().addLast(new ProtobufDecoder(Example.BaseData.getDefaultInstance(), registry));

//encoded

ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());

ch.pipeline().addLast(new ProtobufEncoder());

// 注冊handler

ch.pipeline().addLast(new ProtoBufServerHandler());

}

});

//綁定端口 同步等待成功

ChannelFuture f = b.bind(port).sync();

//等待服務(wù)端監(jiān)聽端口關(guān)閉

f.channel().closeFuture().sync();

} finally {

workerGroup.shutdownGracefully();

bossGroup.shutdownGracefully();

}

}

public static void main(String[] args) throws Exception {

logger.log(Level.INFO, "NettyServer start...");

new NettyServer().start(PORT);

}

}

ProtoBufServerHandler

package example2.server.handler;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.SimpleChannelInboundHandler;

import java.util.HashMap;

import java.util.Map;

import example2.handlers.AbstractHandler;

import example2.handlers.EquipHandler;

import example2.handlers.FriendHandler;

import example2.proto.Example;

import example2.proto.Example.BaseData;

import example2.proto.Example.Header;

public class ProtoBufServerHandler extends SimpleChannelInboundHandler<Example.BaseData> {

//不同類型處理器，應(yīng)該在服務(wù)器啟動的時候就加載好對應(yīng)關(guān)系

private static Map<Example.Header, AbstractHandler> headersMap;

static {

headersMap = new HashMap<Example.Header, AbstractHandler>();

headersMap.put(Header.Msg1001, new EquipHandler());

headersMap.put(Header.Msg1002, new EquipHandler());

headersMap.put(Header.Msg1003, new FriendHandler());

}

@Override

protected void messageReceived(ChannelHandlerContext ctx, BaseData baseData) throws Exception {

//需要放到單獨的分發(fā)器中處理

AbstractHandler abstractHandler = headersMap.get(baseData.getHeader());

if (abstractHandler == null) {

System.err.println("沒有找到消息處理器??！");

} else {

abstractHandler.handleMsg(ctx, baseData);

}

}

}

抽象處理器AbstractHandler

package example2.handlers;

import io.netty.channel.ChannelHandlerContext;

import example2.proto.Example.BaseData;

public abstract class AbstractHandler {

public void handleMsg(ChannelHandlerContext ctx, BaseData baseData) {

try {

Object object = handle(baseData);

ctx.channel().writeAndFlush(object);

} catch (Exception e) {

e.printStackTrace();

}

}

/**

* 處理消息

* 

* @param messageVo

*/

public abstract Object handle(BaseData baseData) throws Exception;

}

FriendHandler處理器

package example2.handlers;

import com.google.protobuf.InvalidProtocolBufferException;

import example2.proto.Example;

import example2.proto.Example.BaseData;

import example2.proto.Friend;

public class FriendHandler extends AbstractHandler {

@Override

public Object handle(BaseData baseData) throws InvalidProtocolBufferException {

switch (baseData.getHeader()) {

case Msg1003:

return addFriend(baseData);

}

return null;

}

private Object addFriend(BaseData baseData) {

Friend.Receive1003 extension = baseData.getExtension(Friend.receive1003);

System.err.println("1003消息接收成功，我要返回消息了----" + extension.getFriendId());

Friend.Send1003.Builder sendMsg = Friend.Send1003.newBuilder();

sendMsg.setFriendId(extension.getFriendId()).setState(0);

Example.BaseData.Builder builder = Example.BaseData.newBuilder();

builder.setHeader(Example.Header.Msg1003);

builder.setExtension(Friend.send1003, sendMsg.build());

return builder.build();

}

}

客戶端跟服務(wù)器差不多的，就不上代碼了。

這樣一套消息處理，如果消息很多，感覺加起來會比較麻煩，大神給指點下怎么樣優(yōu)化可以使功能實現(xiàn)更簡潔，萬分感謝！

--------------------- 

作者：醉從零 

來源：CSDN 

原文：https://blog.csdn.net/woshiicesky/article/details/78044535 

版權(quán)聲明：本文為博主原創(chuàng)文章，轉(zhuǎn)載請附上博文鏈接！