這里解決了三個(gè)問題
- 協(xié)議定義,解決 粘包/拆包 問題
- 單客戶端并發(fā)發(fā)送/消息維護(hù)問題
- 服務(wù)端并發(fā)提供服務(wù)問題
三個(gè)問題的具體實(shí)現(xiàn)如下
1.協(xié)議定義:
完整數(shù)據(jù)塊包含數(shù)據(jù) 開始標(biāo)識頭,數(shù)據(jù)長度,真實(shí)數(shù)據(jù)三部分,如下圖.
在這里插入圖片描述
客戶端,具體發(fā)送代碼實(shí)現(xiàn)如下:
public class RpcEncoder extends MessageToByteEncoder {
@Override
protected void encode(ChannelHandlerContext ctx, Object requestBoday, ByteBuf out) throws Exception {
//序列化傳輸對象. 也可只是傳輸字符串,服務(wù)端解析,但是局限不較大,無法應(yīng)對多樣的調(diào)用函數(shù),對應(yīng)參數(shù),已經(jīng)類型
byte[] data = SerializationUtil.serialize(requestBoday);
//先寫入 開始標(biāo)識
out.writeBytes(Constants.SERVIE_HEARD.getBytes());
//再寫入數(shù)據(jù)長度
out.writeInt(data.length);
//再寫入真實(shí)數(shù)據(jù)
out.writeBytes(data);
}
}
服務(wù)端,具體接收解析代碼實(shí)現(xiàn)如下:
public class RpcDecoder extends ByteToMessageDecoder {
.............
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
//hadReadHeard避免多次判斷頭信息
if (!hadReadHeard) {
while (true) {
//這里保證至少讀到一個(gè)頭信息，也可以讀到一個(gè)頭和數(shù)據(jù)長度在做處理
if (in.readableBytes() < 4) {
return;
}
in.markReaderIndex();
in.readBytes(dataHeardBuffer);
System.out.println(Constants.SERVIE_HEARD.getBytes().length);
String s = new String(dataHeardBuffer);
//讀到頭標(biāo)識信息,準(zhǔn)備讀取數(shù)據(jù)長度和數(shù)據(jù)
if (s.equals(Constants.SERVIE_HEARD)) {
hadReadHeard = true;
break;
} else {
in.resetReaderIndex();
//為讀取到 頭標(biāo)識,則過濾一個(gè)字節(jié),繼續(xù)判斷是否收到頭標(biāo)識
in.readByte();
}
}
}
in.markReaderIndex();
int dataLength = in.readInt();
if (in.readableBytes() < dataLength) {
in.resetReaderIndex();
return;
}
hadReadHeard = false;
byte[] data = new byte[dataLength];
in.readBytes(data);
out.add(SerializationUtil.deserialize(data, requestResponseRpc));
}
}
2.單客戶端并發(fā)發(fā)送/消息維護(hù)問題:
發(fā)送消息的維護(hù):
1)消息通過唯一id來區(qū)分
2)所有"發(fā)送的消息" 都記錄到hashmap中維護(hù)記錄.
3)發(fā)送消息后,會阻塞等待結(jié)果返回
4)所有接收的消息,都借助唯一ID匹配到"發(fā)送的消息",并喚醒(notify)阻塞的發(fā)送線程處理返回?cái)?shù)據(jù)
public class ProxyHelperTool {
...........
public <T> T create(final Class<?> interfaceClass) {
return (T) Proxy.newProxyInstance(
interfaceClass.getClassLoader(),
new Class<?>[]{interfaceClass},
new InvocationHandler() {
//@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
if (method.getDeclaringClass().getAnnotation(ServiceName.class) == null) {
throw new RuntimeException("Annotation(ServiceName) is null.");
}
//構(gòu)造請求消息,并獲取請求服務(wù),方法,參數(shù),參數(shù)類型
RequestRpc requestRpc = new RequestRpc();
requestRpc.setMethodName(method.getName());
requestRpc.setServiceName(method.getDeclaringClass().getAnnotation(ServiceName.class).name());
requestRpc.setParameters(args);
requestRpc.setParameterTypes(method.getParameterTypes());
//設(shè)置唯一id,確保消息的唯一性
requestRpc.setRequestId(StringUtil.getUiid());
//將發(fā)送的消息 送入列表維護(hù)起來.
ClientHandler.waitingRPC.put(requestRpc.getRequestId(),requestRpc);
ProxyHelperTool.client.send(requestRpc);
//進(jìn)入阻塞等待,直到服務(wù)返回消息 喚醒.To do:這里缺過時(shí)處理
synchronized(requestRpc){
requestRpc.wait();
}
return requestRpc.getResult();
}
}
);
}
}
3.服務(wù)端并發(fā)服務(wù):
public class ServerHandler extends ChannelInboundHandlerAdapter {
.............
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//將服務(wù)方靜線程里執(zhí)行,避免阻塞
ServerService.submit(new Runnable() {
@Override
public void run() {
RequestRpc requestRpc = (RequestRpc)msg;
ResponseRpc responseRpc = handle(requestRpc);
responseRpc.setRequestId(requestRpc.getRequestId());
ctx.writeAndFlush(responseRpc).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
System.out.println("Server operationComplete");
}
});
}
});
/*.addListener(ChannelFutureListener.CLOSE)*/
}
//真實(shí)處理服務(wù)的地方,依據(jù)對方傳遞的 調(diào)用服務(wù)和參數(shù)通過反射調(diào)用獲取結(jié)果返回
private ResponseRpc handle(RequestRpc requestRpc){
ResponseRpc responseRpc = new ResponseRpc();
Object object = ServerService.getService(requestRpc.getServiceName());
if(object == null){
responseRpc.setException(new RuntimeException("Not service:"+requestRpc.
getServiceName()));
return responseRpc;
}
try {
Class<?> serviceClass = object.getClass();
Method method = serviceClass.getMethod(requestRpc.getMethodName(),
requestRpc.getParameterTypes());
method.setAccessible(true);
Object[] parameters = requestRpc.getParameters();
responseRpc.setResult(method.invoke(object, parameters));
} catch (Exception e){
responseRpc.setResult(e);
}
return responseRpc;
}
........
}
測試方式,以及結(jié)果
客戶端 測試模擬 調(diào)用遠(yuǎn)程服務(wù)
這里, 客戶端建立單鏈接,并發(fā)發(fā)送消息的方式 向服務(wù)端發(fā)起服務(wù)調(diào)用
public class TestClient {
public static ProxyHelperTool proxyHelperTool = new ProxyHelperTool();
public static void main(String[] args) throws Exception {
int threadNumber = 15;
CountDownLatch countDownLatch = new CountDownLatch(threadNumber);
//開始15個(gè)線程發(fā)送 服務(wù)調(diào)用消息
for(int i=0;i<threadNumber;i++){
new Thread(){
@Override
public void run() {
//客戶端,通過傳遞當(dāng)前線程的名稱（Thread.currentThread().getName）給服務(wù)端；
//服務(wù)端,組合收到的字符 再次發(fā)回來。
//通過對比 "線程名"，可見各個(gè)線程收到的是否是自己發(fā)送的。
MsgService msgService = proxyHelperTool.create(MsgService.class);
String reslut = msgService.send(Thread.currentThread().getName());
System.out.println("Client("+Thread.currentThread().getName()+") get mag:" + "\n" + "..." + reslut);
countDownLatch.countDown();
}
}.start();
}
countDownLatch.await();
ClientHelper.getClientHelper().close();
}
}
客戶端 測試模擬 收到的結(jié)果
可見對應(yīng)的調(diào)用線程,都收到了自己發(fā)出去的消息. 對應(yīng)的thread-name 匹配
在這里插入圖片描述
參考
https://my.oschina.net/huangyong/blog/361751?fromerr=NpC3phqY
https://github.com/apache/hadoop
