rabbitmq实现高吞吐量的rpc调用

rabbitmq实现rpc调用基本思路：
客户端（client）：客户端发起rpc调用，这当成一个消息，发送到rabbitmq服务器。这个消息会携带两个特殊（额外）的信息，一个是调用序号，一个是回调队列名称。调用序号需要服务端原样返回，而回调队列名称是用于服务端将结果放入这个队列中，以便客户端取回结果。
服务端（service）：服务端接收到了一个rpc调用后，执行调用代码，将结果返回到指定的回调队列中去。
这里我们可以约定一个数据模型，见代码：
public class RpcInvokeModel implements Serializable {
          //真实传输的数据
private Object target;
          //调用序号
private Long invokeFlag;
          //回调队列名称
        private String callBackQueue;
}
客户端将所有的调用封装成这个模型，然后发送出去。服务端解析成这个模型，当然，你可以再这个模型中加入别的参数（如方法名称等等）。
客户端代码：
package com.pdy.rabbitmq.rpc;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
* Represents a connection with a queue
*
* @author syntx
*
*/
public abstract class ClientEndPoint {

protected Channel producerChannel;
protected Channel consumerChannel;
protected Connection connection;
protected String endPointName;

public ClientEndPoint(String endpointName) throws IOException,
TimeoutException {
this.endPointName = endpointName;

// Create a connection factory
ConnectionFactory factory = new ConnectionFactory();

// hostname of your rabbitmq server
// factory.setVirtualHost("pdy");
factory.setHost("localhost");
// factory.setUsername("pdy");
// factory.setPassword("pdy");
// factory.setVirtualHost("pdy11");
// getting a connection
connection = factory.newConnection();
// creating a channel
producerChannel = connection.createChannel();
consumerChannel = connection.createChannel();

producerChannel.queueDeclare(endpointName, false, false, false, null);
}

/**
* 关闭channel和connection。并非必须，因为隐含是自动调用的。
*
* @throws IOException
* @throws TimeoutException
*/
public void close() throws IOException, TimeoutException {
this.producerChannel.close();
this.consumerChannel.close();
this.connection.close();
}
}

package com.pdy.rabbitmq.rpc;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeoutException;

import org.apache.commons.lang3.SerializationUtils;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.AMQP.Queue.DeclareOk;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;

public class Client extends ClientEndPoint implements Consumer, Runnable {

public Client(String endpointName) throws IOException, TimeoutException {
super(endpointName);
DeclareOk ok = super.consumerChannel.queueDeclare();
callBackQueue = ok.getQueue();
super.consumerChannel.basicConsume(callBackQueue, true, this);
}

@Override
public void handleConsumeOk(String consumerTag) {

}

@Override
public void handleCancelOk(String consumerTag) {

}

@Override
public void handleCancel(String consumerTag) throws IOException {

}

@Override
public void handleDelivery(String arg0, Envelope arg1,
BasicProperties arg2, byte[] arg3) throws IOException {

RpcInvokeModel invokeModel = SerializationUtils.deserialize(arg3);

synchronized (callResult) {

callResult.put(invokeModel.getInvokeFlag(), invokeModel.getObj());
callResult.notifyAll();
}

}

@Override
public void handleShutdownSignal(String consumerTag,
ShutdownSignalException sig) {

}

@Override
public void handleRecoverOk(String consumerTag) {

}

public void start() throws IOException {

Random random = new Random();
for (int i = 0; i < 10; i++) {

Double[] arr = new Double[10];

for (int j = 0; j < arr.length; j++) {
arr[j] = random.nextDouble();
}

System.out.println("调用rpc服务：" + Arrays.toString(arr));

RpcFuture<Double[]> result = invokeSort(arr);
Double[] obj = result.get();
System.out.println("调用rpc服务响应的结果：" + Arrays.toString(obj));
}
}

private RpcFuture<Double[]> invokeSort(Double[] arr) throws IOException {

RpcInvokeModel mo = new RpcInvokeModel();
mo.setInvokeFlag(++invokeFlag);
mo.setObj(arr);
byte[] body = SerializationUtils.serialize(mo);

BasicProperties basicProperties = new BasicProperties().builder()
.replyTo(callBackQueue).build();
super.producerChannel.basicPublish("", super.endPointName,
basicProperties, body);

return new RpcFuture<>(this, mo.getInvokeFlag());
}

private final String callBackQueue;
private final Map<Long, Object> callResult = new HashMap<>();
private volatile long invokeFlag = 0;

public Object getResultByFlagKey(Long flagKey) {

Object result = null;
while (true) {
synchronized (callResult) {
result = callResult.remove(flagKey);
if (result == null) {
try {
callResult.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
return result;
}
}
}
}

@Override
public void run() {

try {
start();
} catch (IOException e) {
e.printStackTrace();
}
}
}

package com.pdy.rabbitmq.rpc;

public class RpcFuture<T> {

private final Client client;
private final Long flagKey;

public RpcFuture(Client client, Long flagKey) {

this.client = client;
this.flagKey = flagKey;
}

public T get() {

return (T) client.getResultByFlagKey(flagKey);
}
}


服务端代码：
package com.pdy.rabbitmq.rpc;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
* Represents a connection with a queue
*
* @author syntx
*
*/
public abstract class ServiceEndPoint {

protected Channel producerChannel;
protected Channel consumerChannel;
protected Connection connection;
protected String endPointName;

public ServiceEndPoint(String endpointName) throws IOException,
TimeoutException {
this.endPointName = endpointName;

// Create a connection factory
ConnectionFactory factory = new ConnectionFactory();

// hostname of your rabbitmq server
// factory.setVirtualHost("pdy");
factory.setHost("localhost");
// factory.setUsername("pdy");
// factory.setPassword("pdy");
// factory.setVirtualHost("pdy11");
// getting a connection
connection = factory.newConnection();
// creating a channel
producerChannel = connection.createChannel();
consumerChannel = connection.createChannel();

consumerChannel.queueDeclare(endpointName, false, false, false, null);
}

/**
* 关闭channel和connection。并非必须，因为隐含是自动调用的。
*
* @throws IOException
* @throws TimeoutException
*/
public void close() throws IOException, TimeoutException {
this.producerChannel.close();
this.consumerChannel.close();
this.connection.close();
}
}

package com.pdy.rabbitmq.rpc;

import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.TimeoutException;

import org.apache.commons.lang3.SerializationUtils;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;

public class Service extends ServiceEndPoint implements Consumer {

public Service(String endpointName) throws IOException, TimeoutException {
super(endpointName);
}

public void start() throws IOException {

super.consumerChannel.basicConsume(super.endPointName, true, this);
}

@Override
public void handleConsumeOk(String consumerTag) {

}

@Override
public void handleCancelOk(String consumerTag) {

}

@Override
public void handleCancel(String consumerTag) throws IOException {

}

@Override
public void handleDelivery(String arg0, Envelope arg1,
BasicProperties arg2, byte[] arg3) throws IOException {

RpcInvokeModel invokeModel = SerializationUtils.deserialize(arg3);
Double[] obj = (Double[]) invokeModel.getObj();

Arrays.sort(obj);

invokeModel.setObj(obj);

byte[] body = SerializationUtils.serialize(invokeModel);

String routingKey = arg2.getReplyTo();
super.producerChannel.basicPublish("", routingKey, null, body);
}

@Override
public void handleShutdownSignal(String consumerTag,
ShutdownSignalException sig) {

}

@Override
public void handleRecoverOk(String consumerTag) {

}
}

测试代码：
package test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.pdy.rabbitmq.rpc.Client;
import com.pdy.rabbitmq.rpc.Service;

public class RpcTest {

public static void main(String[] args) throws IOException, TimeoutException {

String queue = "pdyRpc";
client(queue);
service(queue);
}

private static void service(String queue) throws IOException,
TimeoutException {

Service service = new Service(queue);
service.start();
}

private static void client(String queue) throws IOException,
TimeoutException {

Client client = new Client(queue);
new Thread(client).start();
}
}

代码还不满足线程安全，但是改成线程安全是很容易的事情。