springboot-MQRPCRPC 調(diào)用框架

一個簡單便捷的基于springboot+RabbitMQ中間件實現(xiàn)的RPC調(diào)用框架

遠程調(diào)用過程如下

首先：消費者和生產(chǎn)者spring容器初始化的時候，會根據(jù)配置的的api在RabbitMQ上建立相應(yīng)的隊列,消費者會監(jiān)聽相關(guān)隊列

1）生產(chǎn)者（client）調(diào)用以本地調(diào)用方式調(diào)用服務(wù)；

2）client 接收到調(diào)用后通過Hessian將方法、參數(shù)等組裝成能夠進行網(wǎng)絡(luò)傳輸?shù)南Ⅲw；

3）client 通過代理類，執(zhí)行invoke方法，統(tǒng)一將消息發(fā)送到MQ監(jiān)聽的服務(wù)端；

4）server 收到消息后通過Hessian進行解碼；

5）server 根據(jù)解碼結(jié)果調(diào)用本地的服務(wù)；

6）本地服務(wù)執(zhí)行并將結(jié)果返回給server ；

7）server 將返回結(jié)果通過Hessian打包發(fā)送至消費方；

8）client 接收到消息，并進行解碼；

9）生產(chǎn)者得到最終結(jié)果。

客戶端配置：

    <!-- RabbitMQ連接池 -->
    <task:executor id="amqpConnectionTaskExecutor" pool-size="5"/>

    <!-- RabbitMQ連接器 -->
    <rabbit:connection-factory id="connectionFactory" executor="amqpConnectionTaskExecutor" host="127.0.0.1" port="5672" username="admin" password="admin"
                               virtual-host="/kl"/>

    <bean id="myService" class="com.kl.client.MQClientProxyFactoryBean">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="serviceInterface" value="com.kl.api.Service"/>
    </bean>

   <!-- RabbitMQ連接池 -->
    <task:executor id="amqpConnectionTaskExecutor" pool-size="5"/>

    <!-- RabbitMQ連接器 -->
    <rabbit:connection-factory id="connectionFactory" executor="amqpConnectionTaskExecutor" host="127.0.0.1" port="5672" username="admin" password="admin"
                               virtual-host="/kl"/>
    <bean id="ServiceImpl" class="com.kl.apiImpl.ServiceImpl"/>

    <bean id="MyServiceEndpoint" class="com.kl.server.MQServerEndpoint">
        <constructor-arg index="0" ref="ServiceImpl"/>
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>

測試代碼及圖例

    /**
     * 暴力測試
     * @param args
     */
    public static void main(String[] args) {
        GenericXmlApplicationContext context = new GenericXmlApplicationContext(
                "classpath:/applicationContext-client.xml");
        Service service = (Service) context.getBean("myService");
        new ClientTest().exec(service);
    }
    public void exec(Service service){
        ExecutorService executorService= Executors.newFixedThreadPool(30);
        for(int i=0;i<=30;i++){
            executorService.submit(new Task(service));
        }
    }
    private class Task implements Callable {
        private Service service;
        public Task(Service service){
            this.service=service;
        }
        @Override
        public Object call() throws Exception {
            for(int i=0;i<=100000;i++){
                System.out.println("servicEcho當前線程："+Thread.currentThread().getName()+"| 線程任務(wù)數(shù)"+i+"| 輸出："+service.echo("Hello AMQP!"));
                System.out.println("serviceStudent當前線程："+Thread.currentThread().getName()+"| 線程任務(wù)數(shù)"+i+"| 輸出："+service.getStudent(null).getName());

            }
            return null;
        }
    }

以下為測試圖例