隨著復(fù)賽今天截止,為期兩個月的挑戰(zhàn)賽也終于結(jié)束了.這兩個月里很大一部分時間花在這上面,有過歡樂,也有為分?jǐn)?shù)刷不上去而發(fā)愁.作為第一次參加比賽,對比賽結(jié)果還算是滿意吧.而在這個過程中,對多線程知識,netty,nio等知識的深入認(rèn)識.下面是對比賽的總結(jié)和思考.排名如下
初賽：《Service Mesh Agent for Apache Dubbo (Incubating) 》
- 賽題的思考
題目看起來是讓我們實(shí)現(xiàn)一個rpc agent.因?yàn)楣俜揭呀?jīng)給出了consumer和provider,選手就是要實(shí)現(xiàn)兩個代理,第一個代理是consumer-agent,負(fù)責(zé)把consumer的調(diào)用通過自定義協(xié)議發(fā)動給Provider-agent.第二個代理就是provider-agent.他的任務(wù)就是接收Consumer-agent通過網(wǎng)絡(luò)發(fā)動過來的消息,然后通過dubbo調(diào)用provider.最后把結(jié)果返回給consumer-agent.整個系統(tǒng)的調(diào)用圖如下:
- 設(shè)計和實(shí)現(xiàn).
整個調(diào)用過程如下所示:
- ①處這里采用Netty Http應(yīng)用作為服務(wù)端,處理Consumer發(fā)送過來的http請求.
- ②處這里就是在Consumer-agent開啟Netty Client,Provider-agent端開啟Netty Server進(jìn)行請求和響應(yīng).
- ③ Provider-agent通Netty Client去調(diào)用Provider的服務(wù).
- ④ Provider把結(jié)果返回給Consumer-agent.
- ⑤ Consumer-agent把結(jié)果封裝成HttpResponse返回給客戶端.
Provider提供的服務(wù)如下:
public interface IHelloService {
/**
* 計算傳入?yún)?shù)的哈希值.
*
* @param str 隨機(jī)字符串
* @return 該字符串的哈希值
*/
int hash(String str);
}
整個代碼我放在github中,這里不對整個代碼做分析,只分析出關(guān)鍵的點(diǎn).
負(fù)載均衡
如下圖,3個provider的負(fù)載能力如下,那么我們可以選擇負(fù)載均衡算法的時候,把這個考慮進(jìn)去.我選擇是隨機(jī)加權(quán)算法.根據(jù)大家的一致認(rèn)同,small:meddium:large = 1:2:2.
所有的服務(wù)都運(yùn)行在docker環(huán)境中,而用的etcd作為服務(wù)發(fā)現(xiàn)的組件.事先并不知道那臺機(jī)器是small,large,meddium.那么我們可以考慮把參數(shù)加上啟動參數(shù).一旦服務(wù)啟動,這些信息,都會注冊到etcd中.然后取出來,做相應(yīng)的判斷就行.
在etcd做服務(wù)發(fā)現(xiàn)的時候,把型號信息轉(zhuǎn)換成比例注冊上去
//small 1; meddium和large是2.
if(val.equals("small")) {
endpoints.add(new Endpoint(host, port, 1));
}else{
endpoints.add(new Endpoint(host, port, 2));
}
Consumer在選擇那個Provider的時候就可以根據(jù)以上的信息,輪詢選擇一個.
//向endpoints加入5個實(shí)例,small一個,meddium和large都是2個.
if (null == endpoints) {
synchronized (ConsumerAgentHttpServerHandler.class) {
if (null == endpoints) {
endpoints = RegistryInstance.getInstance().find("com.alibaba.dubbo.performance.demo.provider.IHelloService");
ListIterator<Endpoint> it = endpoints.listIterator();
while (it.hasNext()){
Endpoint temp = it.next();
if(temp.getSize()==2) {
it.add(temp);
}
}
}
}
}
int id = count.getAndIncrement();
if(id>=4){
count.set(0);
id=4;
}
// 簡單的負(fù)載均衡，隨機(jī)取一個
Endpoint endpoint = endpoints.get(id);
這樣一個隨機(jī)加權(quán)的算法就實(shí)現(xiàn)了.
EventLoop復(fù)用
當(dāng)我們創(chuàng)建Provident-agent的時候,我們是否可以考慮Eventloop的復(fù)用,這樣每個請求從接收到發(fā)動都是用同一個線程處理的,沒有上下文切換.另外一個,這樣做好處,把channel和Eventloop綁定起來,也就限定了channel的個數(shù),相當(dāng)于做了一個channel的緩存(因?yàn)閏hannel的數(shù)量得控制).一舉兩得.
private void providerServerStart(int port) {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup(4);
putMap(workerGroup);
try {
ServerBootstrap sbs = new ServerBootstrap().group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.localAddress(new InetSocketAddress(port))
.childHandler(new ProviderAgentHttpServerChannelInitializer());
LOGGER.info("provider netty server start");
ChannelFuture future = sbs.bind(port).sync();
future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
//預(yù)先把channel設(shè)置好,復(fù)用上面的eventloop.
public void putMap(EventLoopGroup group) {
for (EventExecutor executor : group) {
try {
map.put((EventLoop) executor, connecManager.getChannel( (EventLoop) executor));
} catch (Exception e) {
e.printStackTrace();
}
}
}
回調(diào)的設(shè)計
當(dāng)Provider返回給結(jié)果后,那我們應(yīng)該如何把結(jié)果返回給Consumer-agent呢,也就是它如何記住之前的通道.這里采用的是一個回調(diào)的設(shè)計.這樣就能夠記住上下文,也就是記住過來時候的ChannelHandlerContext,通過這個把結(jié)果返回回去.
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
Map<String, String> data = HttpParser.parse(msg);
handle(new RequestWrapper(data.get("interface"),
data.get("method"),
data.get("parameterTypesString"),
data.get("parameter")), (result) -> {
FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(result.getBytes()));
response.headers().set(CONTENT_TYPE, "text/plain");
response.headers().setInt(CONTENT_LENGTH, response.content().readableBytes());
ctx.write(response);
ctx.writeAndFlush(Unpooled.EMPTY_BUFFER);
},ctx.channel().eventLoop());
}
當(dāng)結(jié)果返回后,通過回調(diào)調(diào)用回調(diào)函數(shù)的邏輯
//拿到結(jié)果后,回調(diào)
public void done(RpcResponse response){
this.response = response;
sender.accept(new String(response.getBytes()).trim());
}
- 反思和思考
可以批量flush,批量decode(來源于朋友徐靖峰的思想)
Netty 提供了一個方便的解碼工具類 ByteToMessageDecoder 嘶朱，如圖上半部分所示配并，這個類具備 accumulate 批量解包能力，可以盡可能的從 socket 里讀取字節(jié)，然后同步調(diào)用 decode 方法提揍，解碼出業(yè)務(wù)對象啤月，并組成一個 List 。最后再循環(huán)遍歷該 List 劳跃，依次提交到 ChannelPipeline 進(jìn)行處理谎仲。此處我們做了一個細(xì)小的改動，如圖下半部分所示刨仑，即將提交的內(nèi)容從單個 command 郑诺，改為整個 List 一起提交，如此能減少 pipeline 的執(zhí)行次數(shù)杉武，同時提升吞吐量辙诞。這個模式在低并發(fā)場景，并沒有什么優(yōu)勢轻抱，而在高并發(fā)場景下對提升吞吐量有不小的性能提升飞涂。
負(fù)載均衡
上面我的做法有點(diǎn)硬編碼的意思,而且隨機(jī)的話,而且不確定性有點(diǎn)大.那是是否可以考慮根據(jù)調(diào)用的次數(shù)來做負(fù)載均衡,也就是說,給句每個Provider請求的次數(shù),盡量把請求分給請求量少的Provider,當(dāng)然這個量還是得加權(quán).實(shí)現(xiàn)的復(fù)雜性有點(diǎn)高.
限流
經(jīng)過朋友提醒,是否可以嘗試下,限流,也就是說不放那么多請求進(jìn)取,只通過一部分來請求,待完成之后,再放另外一部分,這個可以嘗試用令牌桶來實(shí)現(xiàn).處于理論階段,沒實(shí)際嘗試過.
編碼
我做的處理里面都是采用的jdk自帶的編碼方式.如果采用kryo,protobuf的方式,性能上也會有一定的提升.
我的代碼:https://github.com/maskwang520/springforall.git
復(fù)賽：實(shí)現(xiàn)一個進(jìn)程內(nèi)的隊(duì)列引擎，單機(jī)可支持100萬隊(duì)列以上,能夠承受2億消息的存取.
- 賽題的思考
題目要求有5個:
1.各個階段線程數(shù)在20~30左右
2.發(fā)送階段：消息大小在50字節(jié)左右祈搜，消息條數(shù)在20億條左右较店，也即發(fā)送總數(shù)據(jù)在100G左右
3.索引校驗(yàn)階段：會對所有隊(duì)列的索引進(jìn)行隨機(jī)校驗(yàn)；平均每個隊(duì)列會校驗(yàn)1~2次容燕；
4.順序消費(fèi)階段：挑選20%的隊(duì)列進(jìn)行全部讀取和校驗(yàn)泽西；
5.發(fā)送階段最大耗時不能超過1800s；索引校驗(yàn)階段和順序消費(fèi)階段加在一起缰趋，最大耗時也不能超過1800s捧杉；超時會被判斷為評測失敗。
100萬個queue,20億消息,如果放內(nèi)存是完全不現(xiàn)實(shí)的,內(nèi)存肯定會爆.接下來自然想到把消息存放到文件中,內(nèi)存中只放索引就行.但是內(nèi)存存放索引,是20億消息的消息,索引自然是由(消息起始位置+長度)構(gòu)成.但是這樣的Map<queue,Index>存放的索引有20億,瘋狂的FullGc是不可避免的,Full Gc一多,Tps自然上不去.后來想到,消息按塊存儲(多個消息存在一個塊中),索引的時候按塊索引.這樣就能把Map里面存的只有100萬(queue的個數(shù)),示意圖如下:
Block的設(shè)計
public class Block {
//開始位置
public long startPosition;
//長度
public int length;
//Block中已經(jīng)存放的消息的條數(shù)
public int size;
public Block(Long startPosition, int length) {
this.startPosition = startPosition;
this.length = length;
this.size = 0;
}
}
因?yàn)橐粋€queue中可能有多個Block,在消息檢索的時候給出的是在隊(duì)列中的偏移量,那么size這個域方便后面消息檢索的時候判斷在哪個block中.
消息緩存的設(shè)計
因?yàn)槊慨?dāng)來一個消息都要flush到文件中去,這樣Io的時間就太多了,題目的關(guān)鍵點(diǎn)在于如何減少Io的時間.所以可以采用消息的緩存來處理.每當(dāng)來一個消息,就放入緩存中,當(dāng)緩存中超過10次消息的時候,就同步寫入到文件中去.這樣的話,相當(dāng)于每10次寫,才做一次Io.
public class DataCache {
//消息緩存
public ByteBuffer dataBuffer = ByteBuffer.allocate(1024);
public int count;
}
這里將緩存的大小設(shè)置為1024Byte,當(dāng)然你也可以設(shè)置成更大.這里有個小Tips.緩存的消息最好設(shè)置成Block的大小.這樣當(dāng)緩存滿了之后,就可以直接寫入到一個Block塊中,而不用接著上一個Block寫(上面一個Block寫),這樣設(shè)計,寫入更簡單,每次flush到文件的時候,只要新開辟一個新的Block,而不用管之前的Block.
//以塊為索引,一個隊(duì)列可能有多個塊,且塊的寫入有順序,所有用List來存Block.
public Map<String, List<Block>> blockMap = new ConcurrentHashMap<>();
public Map<String, DataCache> cacheMap = new ConcurrentHashMap<>();
消息的存儲
因?yàn)椴豢赡苊總€隊(duì)列的消息都用一個文件來存放,所以這里用hash來把文件限定在32個.一個queue的Block必須在一個文件里面.不同queue的Block可以在一個文件里面.
//根據(jù)隊(duì)列的名字hash到對應(yīng)的文件中,共32個文件
int hashFile(String queueName) {
return queueName.hashCode() & 0x1f;
//return 0;
}
還存在一個問題就是,往一個文件中寫入消息的時候,什么位置寫,因?yàn)榘磯K寫.所以已經(jīng)寫過的塊不能用.只能從新開辟一個塊,塊與塊之間盡可能緊湊.
//block的大小為1024,根據(jù)當(dāng)前文件已經(jīng)存在的寫的位置,找到下一個比該位置大的,且是1024的倍數(shù)
public long getLeastBlockPosition(long length) {
if (length == 0) {
return 0;
}
int initSize = 1 << 10;
int i = 1;
while (i * initSize <= length) {
i++;
}
//定義到可用的塊的第一個位置
return i * initSize;
}
消息存放
這里采用的是原生的filechannel去讀寫.本打算用mmap去寫的,經(jīng)過一位朋友提醒,mmap在這個場景下不合適.原因是不是長期讀寫,寫完就釋放,不是長期的.
public void put(String queueName, byte[] message) {
int hash = hashFile(queueName);
String path = DIRPATH + hash + ".txt";
lock.lock();
//創(chuàng)建文件
File file = new File(path);
if (!file.exists()) {
try {
file.createNewFile();
} catch (Exception e) {
e.printStackTrace();
}
}
if (!blockMap.containsKey(queueName)) {
List<Block> list = new ArrayList();
blockMap.put(queueName, list);
}
if (!cacheMap.containsKey(queueName)) {
DataCache dataCache = new DataCache();
cacheMap.put(queueName, dataCache);
}
DataCache dataCache = cacheMap.get(queueName);
//每10次flush到文件中
if (dataCache.count == 10) {
FileChannel fileChannel = null;
// long fileLength = 0;
try {
fileChannel = new RandomAccessFile(file, "rw").getChannel();
//fileLength = raf.length();
} catch (Exception e) {
e.printStackTrace();
}
long blockPosition;
try {
blockPosition = getLeastBlockPosition(getLeastBlockPosition(fileChannel.size()));
Block block = new Block(blockPosition, dataCache.dataBuffer.position());
block.size = 10;
blockMap.get(queueName).add(block);
dataCache.dataBuffer.flip();
fileChannel.position(blockPosition);
fileChannel.write(dataCache.dataBuffer);
dataCache.dataBuffer.clear();
} catch (Exception e) {
e.printStackTrace();
}finally {
try {
fileChannel.close();
}catch (Exception e){
e.printStackTrace();
}
}
} else {
//放入緩存中
dataCache.dataBuffer.putInt(message.length);
dataCache.dataBuffer.put(message);
dataCache.count++;
}
lock.unlock();
}
消息獲取
消息獲取的思路是根據(jù)隊(duì)列名,找到該隊(duì)列對應(yīng)的List<Block>,然后根據(jù)偏移量,找到屬于哪個block.找到具體的Block后,然后遍歷Block,找到偏移量的開始位置,取相應(yīng)數(shù)量的消息即可.
public Collection<byte[]> get(String queueName, long offset, long num) {
//隊(duì)列不存在
if (!blockMap.containsKey(queueName)) {
return EMPTY;
}
//消息集合
List<byte[]> msgs = new ArrayList();
List<Block> blocks = blockMap.get(queueName);
int hash = hashFile(queueName);
String path = DIRPATH + hash + ".txt";
FileChannel fileChannel = null;
int size = blocks.get(0).size;
int eleNum = 0;
//記錄了目標(biāo)block所在的下標(biāo)
int blockNum = 0;
lock.lock();
try {
fileChannel = new RandomAccessFile(new File(path), "rw").getChannel();
for (int i = 1; i < blocks.size() && size < offset; i++, blockNum++) {
size += blocks.get(i).size;
}
size = size - blocks.get(blockNum).size;
for (int i = blockNum; i < blocks.size(); i++) {
//size+=blocks.get(i).size;
// size-=blocks.get(i).size;
int length = blocks.get(i).length;
MappedByteBuffer buffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, blocks.get(i).startPosition, length);
int sum = 0;
while (sum < length && size < offset) {
int len = buffer.getInt();
sum += 4;
sum += len;
buffer.position(sum);
size++;
}
if (size >= offset) {
while (buffer.position() < length && eleNum <= num) {
int len = buffer.getInt();
byte[] temp = new byte[len];
buffer.get(temp, 0, len);
eleNum++;
msgs.add(temp);
}
if (eleNum > num) {
break;
}
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
fileChannel.close();
} catch (Exception e) {
e.printStackTrace();
}
lock.unlock();
}
return msgs;
}
思考
- 將所有的ByteBuf池化,包括緩存的那部分ByteBuf.通過ThreadLocal,將ByteBuf與線程綁定起來,后面申請Buffer,直接從對應(yīng)的線程里面去申請即可.
- 在寫入的時候,可以不同步寫,實(shí)現(xiàn)異步寫.由一個線程去異步flush到文件里面
- 當(dāng)讀取消息塊達(dá)到臨界點(diǎn)的時候,由單線程申請buffer資源來預(yù)讀后面的消息塊存入,并緩存.
我的代碼:https://github.com/maskwang520/messagequeue.git
