笔记上一篇介绍了，go语言如何使用protobuf 及生成go 的protobuf 文件，具体内容请见上一篇:go 与 protobuf 安装和使用
1. protobuf 文件定义及注意事项// [开始声明]
syntax = "proto3";
//定义protobuf的包名称空间
package message;
// [结束声明]
// [开始 java 选项配置]
option java_package = "xxxx.core.message";
option java_outer_classname = "PbConnInitInfo";
// [结束 java 选项配置]
// [开始 go 选项配置]
option go_package = "./;proto";
// [结束 go 选项配置]
// [开始 消息定义]
message InitInfo {
string sessionName = 1;
string sessionId = 2;
int32 sessionSize = 3;
}注意一定要加上： option go_package = "./;proto";
它表示go 文件生成的具体位置和 go 文件的包名
生成的initInfo.pb.go
// [开始声明]
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.27.1
// protoc v3.19.1
// source: proto/InitInfo.proto
//定义protobuf的包名称空间
package proto
import (
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
// [开始 消息定义]
type InitInfo struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
SessionName string `protobuf:"bytes,1,opt,name=sessionName,proto3" json:"sessionName,omitempty"`
SessionId string `protobuf:"bytes,2,opt,name=sessionId,proto3" json:"sessionId,omitempty"`
SessionSize int32 `protobuf:"varint,3,opt,name=sessionSize,proto3" json:"sessionSize,omitempty"`
}
func (x *InitInfo) Reset() {
*x = InitInfo{}
if protoimpl.UnsafeEnabled {
mi := &file_proto_InitInfo_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *InitInfo) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*InitInfo) ProtoMessage() {}
func (x *InitInfo) ProtoReflect() protoreflect.Message {
mi := &file_proto_InitInfo_proto_msgTypes[0]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use InitInfo.ProtoReflect.Descriptor instead.
func (*InitInfo) Descriptor() ([]byte, []int) {
return file_proto_InitInfo_proto_rawDescGZIP(), []int{0}
}
func (x *InitInfo) GetSessionName() string {
if x != nil {
return x.SessionName
}
return ""
}
func (x *InitInfo) GetSessionId() string {
if x != nil {
return x.SessionId
}
return ""
}
func (x *InitInfo) GetSessionSize() int32 {
if x != nil {
return x.SessionSize
}
return 0
}
var File_proto_InitInfo_proto protoreflect.FileDescriptor
var file_proto_InitInfo_proto_rawDesc = []byte{
0x0a, 0x14, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x49, 0x6e, 0x69, 0x74, 0x49, 0x6e, 0x66, 0x6f,
0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22,
0x6c, 0x0a, 0x08, 0x49, 0x6e, 0x69, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x20, 0x0a, 0x0b, 0x73,
0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
0x52, 0x0b, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1c, 0x0a,
0x09, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
0x52, 0x09, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x20, 0x0a, 0x0b, 0x73,
0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x53, 0x69, 0x7a, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05,
0x52, 0x0b, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x53, 0x69, 0x7a, 0x65, 0x42, 0x36, 0x0a,
0x11, 0x78, 0x78, 0x78, 0x78, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61,
0x67, 0x65, 0x42, 0x0e, 0x50, 0x62, 0x43, 0x6f, 0x6e, 0x6e, 0x49, 0x6e, 0x69, 0x74, 0x49, 0x6e,
0x66, 0x6f, 0x5a, 0x11, 0x2e, 0x2e, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x70, 0x62, 0x3b,
0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
file_proto_InitInfo_proto_rawDescOnce sync.Once
file_proto_InitInfo_proto_rawDescData = file_proto_InitInfo_proto_rawDesc
)
func file_proto_InitInfo_proto_rawDescGZIP() []byte {
file_proto_InitInfo_proto_rawDescOnce.Do(func() {
file_proto_InitInfo_proto_rawDescData = protoimpl.X.CompressGZIP(file_proto_InitInfo_proto_rawDescData)
})
return file_proto_InitInfo_proto_rawDescData
}
var file_proto_InitInfo_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
var file_proto_InitInfo_proto_goTypes = []interface{}{
(*InitInfo)(nil), // 0: message.InitInfo
}
var file_proto_InitInfo_proto_depIdxs = []int32{
0, // [0:0] is the sub-list for method output_type
0, // [0:0] is the sub-list for method input_type
0, // [0:0] is the sub-list for extension type_name
0, // [0:0] is the sub-list for extension extendee
0, // [0:0] is the sub-list for field type_name
}
func init() { file_proto_InitInfo_proto_init() }
func file_proto_InitInfo_proto_init() {
if File_proto_InitInfo_proto != nil {
return
}
if !protoimpl.UnsafeEnabled {
file_proto_InitInfo_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*InitInfo); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_proto_InitInfo_proto_rawDesc,
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_proto_InitInfo_proto_goTypes,
DependencyIndexes: file_proto_InitInfo_proto_depIdxs,
MessageInfos: file_proto_InitInfo_proto_msgTypes,
}.Build()
File_proto_InitInfo_proto = out.File
file_proto_InitInfo_proto_rawDesc = nil
file_proto_InitInfo_proto_goTypes = nil
file_proto_InitInfo_proto_depIdxs = nil
}
public class NettyTCPConnectionAcceptor implements ConnectionAcceptor {
private static final Logger log = LoggerFactory.getLogger(NettyTCPConnectionAcceptor.class);
private ServerBootstrap serverBootstrap;
private EventLoopGroup bossGroup;
private EventLoopGroup workerGroup;
private Channel channel;
private ServerContext serverContext;
private NettyServerConnectionStore serverConnectionStore;
private NettyServerSessionStore serverSessionStore;
private int port;
private Codec codec;
public NettyTCPConnectionAcceptor(final Codec codec, int port, final ServerContext serverContext) {
this.serverContext = serverContext;
this.serverConnectionStore = new NettyServerConnectionStore();
this.serverSessionStore = new NettyServerSessionStore(serverContext.getServerConfig());
this.serverContext.onStoresReady(this.serverConnectionStore, this.serverSessionStore);
this.port = port;
this.codec = codec;
this.bossGroup = new NioEventLoopGroup(1, new ThreadPerTaskExecutor(new DefaultThreadFactory("netty-acceptor")));
this.workerGroup = new NioEventLoopGroup(0, new ThreadPerTaskExecutor(new DefaultThreadFactory("netty-worker")));
this.serverBootstrap = new ServerBootstrap();
((ServerBootstrap)((ServerBootstrap)((ServerBootstrap)this.serverBootstrap.group(this.bossGroup, this.workerGroup).channel(NioServerSocketChannel.class)).handler(new LoggingHandler(LogLevel.INFO))).option(ChannelOption.SO_BACKLOG, 4096)).childOption(ChannelOption.TCP_NODELAY, true).childOption(ChannelOption.SO_SNDBUF, 1048576).childOption(ChannelOption.SO_RCVBUF, 1048576).childHandler(new ChannelInitializer() {
public void initChannel(SocketChannel ch) {
ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));
ch.pipeline().addLast(new ChannelHandler[]{new ProtobufStringDecoder()});
ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));
ch.pipeline().addLast(new ChannelHandler[]{new ProtobufStringEncoder()});
ch.pipeline().addLast(new ChannelHandler[]{new NettyServerHandler(serverContext)});
}
});
}
public void start() {
try {
log.debug("STARTING");
this.channel = this.serverBootstrap.bind(this.port).sync().channel();
} catch (InterruptedException var2) {
log.debug("START_FAILED", var2);
}
}
public void stop() {
this.channel.disconnect();
try {
this.channel.closeFuture().sync();
} catch (InterruptedException var2) {
var2.printStackTrace();
}
this.workerGroup.shutdownGracefully();
this.bossGroup.shutdownGracefully();
}
public void restart() {
this.stop();
this.start();
}
public void setServerContext(final ServerContext serverContext) {
this.serverContext = serverContext;
}
public void setPort(final int port) {
this.port = port;
}
public void setCodec(final Codec codec) {
this.codec = codec;
} package socket
import (
"encoding/binary"
"errors"
"google.golang.org/protobuf/proto"
"log"
"net"
pb "xxx/serialization/protobuf/proto.pb"
)
const (
TCP = "tcp"
RcsSocketAddr = "localhost:7070"
VarintsLen = 4
)
var (
G_transfer *Transfer
)
//初始化连接传输信息
func InitTransfer() {
var (
tcpAddr *net.TCPAddr
conn net.Conn
err error
)
// 解析TCP address
if tcpAddr, err = net.ResolveTCPAddr(TCP, RcsSocketAddr); err != nil {
log.Fatal("tcp resolve ip addr failed #{err}")
}
//与rcs 建立连接
if conn, err = net.DialTCP(TCP, nil, tcpAddr); err != nil {
log.Fatal("connect rcs failed #{err}")
}
log.Println("connect rcs succeed")
G_transfer = &Transfer{
Conn: conn,
}
}
// tcp连接信息结构体
type Transfer struct {
// 连接
Conn net.Conn
// 传输时的缓冲
Buf [1024 * 2]byte
}
// 这个适合于 java netty protobuf 通信，它的底层就是 Base 128 Varints 编码（压缩算法）处理的
func (transfer *Transfer) ReceiveResponse(message *pb.InitInfo) {
// 先读取4个字节
if _, err := transfer.Conn.Read(transfer.Buf[:VarintsLen]); err != nil {
log.Fatal("read proto.InitInfo4 byte failed")
}
// 读取消息长度
mesLen := binary.BigEndian.Uint32(transfer.Buf[:VarintsLen])
// 往下读 mesLen 的长度信息到 buf
readLen, err := transfer.Conn.Read(transfer.Buf[:mesLen])
if readLen != int(mesLen) || err != nil {
log.Fatal("read real #{mesLen} byte failed")
}
// 将byte[] 信息反序列化成 Message
if err := proto.Unmarshal(transfer.Buf[:mesLen], message); err != nil {
log.Fatal("qsh unmarshal message failed")
}
log.Printf("receive server mes : %v", message)
}
// 将上位机的信息发送给server
func (transfer *Transfer) SendMes(message *pb.InitInfo) (err error) {
var (
sendBytes []byte
readLen int
)
// message 序列化
if sendBytes, err = proto.Marshal(message); err != nil {
log.Fatal("sendMes : marshal #{message} failed")
return err
}
// 先获取发送的总字节数
mesLen := uint32(len(sendBytes))
// 将总字节数长度进行 Varints 编码成4个字节的byte
var buf [VarintsLen]byte
binary.BigEndian.PutUint32(buf[:VarintsLen], mesLen)
// 先写入4个字节
if readLen, err = transfer.Conn.Write(buf[:VarintsLen]);readLen != VarintsLen && err != nil {
if readLen == 0 {
log.Fatal("qsh send 4 byte failed, readLen = #{readLen}")
return errors.New("send 4 byte failed, readLen=0")
}
return
}
// 发送消息
if readLen, err = transfer.Conn.Write(sendBytes); err != nil {
log.Fatal("send message failed , readLen = #{readLen}")
}
return
}3.2 测试读写代码：
package socket
import (
"google.golang.org/protobuf/proto"
"log"
pb "xxx/serialization/protobuf/proto.pb"
"testing"
)
func TestInitTransfer2(t *testing.T) {
InitTransfer()
G_transfer.SendMes(&pb.InitInfo{
SessionName: "123",
SessionId: "0001",
SessionSize: int32(10),
})
G_transfer.ReceiveResponse(&pb.InitInfo{})
}欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）