chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed issue #726 , 单链接复用ByteBuffer数组

Browse Source
This commit is contained in:
七锋
2018-07-19 17:55:53 +08:00
parent 5582a38930
commit b19eaeb89a
2 changed files with 37 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
server/src/main/java/com/alibaba/otter/canal/server/netty/NettyUtils.java
View File

@@ -2,8 +2,12 @@ package com.alibaba.otter.canal.server.netty;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.nio.ByteOrder; import java.nio.ByteOrder;
import java.util.ArrayList;
import java.util.List;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers; import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.buffer.CompositeChannelBuffer;
import org.jboss.netty.channel.Channel; import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFutureListener; import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.Channels; import org.jboss.netty.channel.Channels;
@@ -22,6 +26,20 @@ public class NettyUtils {
public static int HEADER_LENGTH = 4; public static int HEADER_LENGTH = 4;
public static Timer hashedWheelTimer = new HashedWheelTimer(); public static Timer hashedWheelTimer = new HashedWheelTimer();
public static void write(Channel channel, ByteBuffer body, ChannelFutureListener channelFutureListner) {
byte[] header = ByteBuffer.allocate(HEADER_LENGTH).order(ByteOrder.BIG_ENDIAN).putInt(body.limit()).array();
List<ChannelBuffer> components = new ArrayList<ChannelBuffer>(2);
components.add(ChannelBuffers.wrappedBuffer(ByteOrder.BIG_ENDIAN, header));
components.add(ChannelBuffers.wrappedBuffer(body));
if (channelFutureListner == null) {
Channels.write(channel, new CompositeChannelBuffer(ByteOrder.BIG_ENDIAN, components));
} else {
Channels.write(channel, new CompositeChannelBuffer(ByteOrder.BIG_ENDIAN, components))
.addListener(channelFutureListner);
}
}
public static void write(Channel channel, byte[] body, ChannelFutureListener channelFutureListner) { public static void write(Channel channel, byte[] body, ChannelFutureListener channelFutureListner) {
byte[] header = ByteBuffer.allocate(HEADER_LENGTH).order(ByteOrder.BIG_ENDIAN).putInt(body.length).array(); byte[] header = ByteBuffer.allocate(HEADER_LENGTH).order(ByteOrder.BIG_ENDIAN).putInt(body.length).array();
if (channelFutureListner == null) { if (channelFutureListner == null) {

21
server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java
View File

@@ -73,7 +73,7 @@ public class SessionHandler extends SimpleChannelHandler {
} }
embeddedServer.subscribe(clientIdentity); embeddedServer.subscribe(clientIdentity);
ctx.setAttachment(clientIdentity);// 设置状态数据 // ctx.setAttachment(clientIdentity);// 设置状态数据
NettyUtils.ack(ctx.getChannel(), null); NettyUtils.ack(ctx.getChannel(), null);
} else { } else {
NettyUtils.error(401, NettyUtils.error(401,
@@ -146,7 +146,19 @@ public class SessionHandler extends SimpleChannelHandler {
size += com.google.protobuf.CodedOutputStream.computeTagSize(5) size += com.google.protobuf.CodedOutputStream.computeTagSize(5)
+ com.google.protobuf.CodedOutputStream.computeRawVarint32Size(messageSize) + com.google.protobuf.CodedOutputStream.computeRawVarint32Size(messageSize)
+ messageSize; + messageSize;
// TODO recyle bytes[] // recyle bytes
// ByteBuffer byteBuffer = (ByteBuffer)
// ctx.getAttachment();
// if (byteBuffer != null && size <=
// byteBuffer.capacity()) {
// byteBuffer.clear();
// } else {
// byteBuffer =
// ByteBuffer.allocate(size).order(ByteOrder.BIG_ENDIAN);
// ctx.setAttachment(byteBuffer);
// }
// CodedOutputStream output =
// CodedOutputStream.newInstance(byteBuffer);
byte[] body = new byte[size]; byte[] body = new byte[size];
CodedOutputStream output = CodedOutputStream.newInstance(body); CodedOutputStream output = CodedOutputStream.newInstance(body);
output.writeEnum(3, PacketType.MESSAGES.getNumber()); output.writeEnum(3, PacketType.MESSAGES.getNumber());
@@ -160,6 +172,11 @@ public class SessionHandler extends SimpleChannelHandler {
} }
output.checkNoSpaceLeft(); output.checkNoSpaceLeft();
NettyUtils.write(ctx.getChannel(), body, null); NettyUtils.write(ctx.getChannel(), body, null);
// output.flush();
// byteBuffer.flip();
// NettyUtils.write(ctx.getChannel(), byteBuffer,
// null);
} else { } else {
Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder(); Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder();
packetBuilder.setType(PacketType.MESSAGES); packetBuilder.setType(PacketType.MESSAGES);