辅助链接
Dubbo系列之 （一）SPI扩展
Dubbo系列之 （二）Registry注册中心-注册（1）
Dubbo系列之 （三）Registry注册中心-注册（2）
Dubbo系列之 （四）服务订阅（1）
Dubbo系列之 （五）服务订阅（2）
Dubbo系列之 （六）服务订阅（3）
Dubbo系列之 （七）链路层那些事（1）
在讲解dubboTCP端的设计时，先了解下一些类的关系图。它们是如何组织在一起的，每个功能又是什么，接着在进一步深入了解其内涵。
类简介
1、Exchangers（交换器工具类） 用来创建TCP服务(bind)和建立客户端连接(connect)辅助类
2、Transporters（数据流传输工具类）用来创建TCP服务(bind)和建立客户端连接(connect)辅助类，Exchangers的底层内容依赖于Transporters，并且Transporters会根据SPI扩展，来适配合适的tcp通讯框架，比如netty,mina等。
3、Exchanger（交换器） 用来创建TCP链接，通过工具类Exchangers完成,该接口是一个SPI扩展，目前唯一仅有就是HeaderExchanger。从名字的含义可以得到，该协议是具有自定义协议头的交换器，所以取名HeaderExchanger。
4、Transporter（数据传输层） 用来创建TCP连接，通过工具类Transporters完成。它也是一个SPI扩展，比如NettyTransporter，MinaTransporter。
5、ExchangeClient (交换器客户端)，Exchanger的connect()方法返回，即建立了TCP连接后，返回的客户端，接着就是通过该客户端与服务端通信，实例有HeaderExchangeClient、LazyConnectExchangeClient、ReferenceCountExchangeClient。之后分别讲解这3个，Exchangers工具类建立的连接客户端是HeaderExchangeClient。
6、ExchangeServer (交换器服务端端) Exchanger的bind()方法返回，即服务端监听的服务端实例，它监听这某个具体的tcp端口。默认实现是HeaderExchangeServer。
7、RemotingServer（远程的TCP服务端），ExchangeServer类也实现了该接口，代表其也是一个远程服务器，具体的实现有NettyServer，由Transporter的bind()方法返回，具体的Transporter返回相应的远程服务端。比如NettyTransporter#bind()返回NettyServer。
8、Client（TCP客户端），ExchangeClient类也实现了该接口，代表其也是一个TCP客户端，具体实现有NettyClient,由Transporter的connect()方法返回，具体的Transporter返回相应的TCP客户端。比如NettyTransporter#connect()返回NettyClient。
9、Channel (通信通道) ，每建立一个TCP链接就相应创建一个Channel。比如Netty建立连接后，就有一个Channel。这里的Channel指的是dubbo自己定义的一个channel。它与netty的channel建立关联，通过NettyChannel类，框架操作的是NettyChannel，而NettyChannel内部持有一个netty的channel对象。
10、HeaderExchangeChannel（交换器Channel，ExchangeChannel属于交换器Channel），它被HeaderExchangeClient客户端所持有，客户端就是通过HeaderExchangeChannel进行通信的，HeaderExchangeChannel内部持有一个具体的Channel。
11、ChannelHandler (通道处理器) 用来处理建立连接、发送请求、结束请求等操作的具体抽象。
12、ChannelHandlers（通道处理器工具类) 主要用来包裹封装具体的Channel,它的作用是通过消息类型，根据Dispatcher返回不同的
13、Dispatcher(消息派发器)
类关系图
试一把，Netty操作--客户端多线程，单链路（TCP）
1、定义传输消息
@Data
@ToString
public class SampleMessage {
private String threadName;
private String id;
private String desc;
}
2、编写编码器
public class SampleEncoder extends MessageToByteEncoder<SampleMessage> {
protected void encode(ChannelHandlerContext channelHandlerContext, SampleMessage sampleMessage, ByteBuf byteBuf) throws Exception {
String threadName = sampleMessage.getThreadName();
String id = sampleMessage.getId();
String desc = sampleMessage.getDesc();
byteBuf.writeInt(threadName.getBytes().length);
byteBuf.writeBytes(threadName.getBytes());
byteBuf.writeInt(id.getBytes().length);
byteBuf.writeBytes(id.getBytes());
byteBuf.writeInt(desc.getBytes().length);
byteBuf.writeBytes(desc.getBytes());
String str = sampleMessage.getThreadName() + ":" + sampleMessage.getDesc() + ":" + sampleMessage.getId();
System.out.println(str);
}
}
3、编写解码器
public class SampleDecoder extends ByteToMessageDecoder {
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
byteBuf.markReaderIndex();
String threadName = read(byteBuf);
if (threadName == null) {
byteBuf.resetReaderIndex();
return;
}
String id = read(byteBuf);
if (id == null) {
byteBuf.resetReaderIndex();
return;
}
String desc = read(byteBuf);
if (desc == null) {
byteBuf.resetReaderIndex();
return;
}
SampleMessage sampleMessage = new SampleMessage();
sampleMessage.setId(id);
sampleMessage.setThreadName(threadName);
sampleMessage.setDesc(desc);
list.add(sampleMessage);
}
private String read(ByteBuf byteBuf) {
if (canReadInt(byteBuf)) {
int readInt = byteBuf.readInt();
if (canReadN(byteBuf, readInt)) {
byte[] bytes = new byte[readInt];
byteBuf.readBytes(bytes);
return new String(bytes);
}
}
return null;
}
private boolean canReadInt(ByteBuf byteBuf) {
return canReadN(byteBuf, 4);
}
private boolean canReadN(ByteBuf byteBuf, int n) {
if (!byteBuf.isReadable()) {
return false;
}
return byteBuf.readableBytes() >= n;
}
}
4、编写消息处理器
public class PrintChannelHandlers extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
if (msg instanceof SampleMessage) {
SampleMessage sampleMessage = (SampleMessage) msg;
System.out.println(sampleMessage.getThreadName() + ":" + sampleMessage.getId() + ":" + sampleMessage.getDesc());
}
}
}
5、编写服务端
public class NettyServerMain {
public static void main(String[] args) {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(new NioEventLoopGroup(1), new NioEventLoopGroup(12))
.channel(NioServerSocketChannel.class)
.childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
// .addLast("log",new LoggingHandler(LogLevel.INFO))
.addLast("decoder", new SampleDecoder())
.addLast("encoder", new SampleEncoder())
.addLast("handler", new PrintChannelHandlers());
}
});
ChannelFuture channelFuture = serverBootstrap.bind(8888);
channelFuture.syncUninterruptibly();
System.out.println("链接前");
Channel channel = channelFuture.channel();
System.out.println("链接后");
}
}
6、编写客户端
public class NettyClientMain {
public static void main(String[] args) {
NettyClientMain nettyClientMain = new NettyClientMain();
nettyClientMain.open();
}
public void open() {
Bootstrap bootstrap = new Bootstrap();
bootstrap = new Bootstrap();
bootstrap.group(new NioEventLoopGroup(10))
.option(ChannelOption.SO_KEEPALIVE, true)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
.channel(NioSocketChannel.class);
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
.addLast("decoder", new SampleDecoder())
.addLast("encoder", new SampleEncoder());
//.addLast("handler", new PrintChannelHandlers());
}
});
SocketAddress socketAddress = new InetSocketAddress("127.0.0.1", 8888);
ChannelFuture future = bootstrap.connect(socketAddress);
boolean ret = future.awaitUninterruptibly(3000, MILLISECONDS);
if (ret && future.isSuccess()) {
Channel newChannel = future.channel();
doProcess(newChannel);
}
}
private void doProcess(Channel channel) {
AtomicLong atomicLong = new AtomicLong();
for (int i = 0; i < 15; i++) {
final char ch = (char) (i + 65);
final String id = "id" + i;
Thread t = new Thread(new Runnable() {
@Override
public void run() {
while (true) {
SampleMessage sampleMessage = new SampleMessage();
sampleMessage.setThreadName(Thread.currentThread().getName());
sampleMessage.setDesc(getdes(ch));
sampleMessage.setId("id" + sampleMessage.getDesc().length() + "-" + atomicLong.getAndIncrement());
channel.writeAndFlush(sampleMessage);
}
}
});
t.start();
}
}
private String getdes(char a) {
Random random = new Random();
StringBuffer buffer = new StringBuffer();
for (int i = 0; i < random.nextInt(500) + 1; i++) {
buffer.append(a);
}
return buffer.toString();
}
}
7、测试结果
结果符合预期，dubbo 也是通过服务底层公用一条TCP链接，多线程进行调用该链路channel。