译: 6. RabbitMQ Spring AMQP 之 RPC

Remote procedure call (RPC)

在第二篇教程中，咱们学习了如何使用工做队列在多个工做人员之间分配耗时的任务。

可是若是咱们须要在远程计算机上运行一个函数并等待结果呢？嗯，这是一个不一样的故事。此模式一般称为远程过程调用或RPC。

在本教程中，咱们将使用RabbitMQ构建RPC系统：客户端和可伸缩的RPC服务器。

因为咱们没有任何值得分发的耗时任务，咱们将建立一个返回Fibonacci数字的虚拟RPC服务。

客户端界面

为了说明如何使用RPC服务，咱们将从“Sender”和“Receiver to”Client“和”Server“更改咱们的配置文件的名称。当咱们调用服务器时，咱们将返回参数的fibonacci咱们打电话给。

Integer response = (Integer) template.convertSendAndReceive
    (exchange.getName(), "rpc", start++);
System.out.println(" [.] Got '" + response + "'");

关于RPC的说明

尽管RPC在计算中是一种很是常见的模式，但它常常受到批评。当程序员不知道函数调用是本地的仍是慢的RPC时，会出现问题。这样的混淆致使系统不可预测，并增长了调试的没必要要的复杂性。错误使用RPC能够致使不可维护的代码，而不是简化软件。

考虑到这一点，请考虑如下建议：

• 确保明显哪一个函数调用是本地的，哪一个是远程的。
• 记录您的系统。使组件之间的依赖关系变得清晰。
• 处理错误案例。当RPC服务器长时间停机时，客户端应该如何反应？

若有疑问，请避免使用RPC。若是能够，您应该使用异步管道 - 而不是相似RPC的阻塞，将结果异步推送到下一个计算阶段。

回调队列

通常来讲，经过RabbitMQ进行RPC很容易。客户端发送请求消息，服务器回复响应消息。为了接收响应，咱们须要发送带有请求的“回调”队列地址。当咱们使用上面的'convertSendAndReceive（）'方法时，Spring-amqp的RabbitTemplate为咱们处理回调队列。使用RabbitTemplate时无需进行任何其余设置。有关详细说明，请参阅请求/回复消息。

消息属性

AMQP 0-9-1协议预约义了一组带有消息的14个属性。大多数属性不多使用，但如下状况除外：

• deliveryMode：将消息标记为持久性（值为2）或瞬态（任何其余值）。您可能会记住第二个教程中的这个属性。
• contentType：用于描述编码的mime类型。例如，对于常用的JSON编码，将此属性设置为：application / json是一种很好的作法。
• replyTo：一般用于命名回调队列。
• correlationId：用于将RPC响应与请求相关联。

相关ID

Spring-amqp容许您专一于您正在使用的消息样式，并隐藏支持此样式所需的消息管道的详细信息。例如，一般本机客户端会为每一个RPC请求建立一个回调队列。这很是低效，所以另外一种方法是为每一个客户端建立一个回调队列。

这引起了一个新问题，在该队列中收到响应后，不清楚响应属于哪一个请求。那是在使用correlationId属性的时候 。Spring-amqp会自动为每一个请求设置一个惟一值。此外，它还处理将响应与正确的correlationID匹配的详细信息。

spring-amqp使rpc样式更容易的一个缘由是，有时您可能但愿忽略回调队列中的未知消息，而不是失败并出现错误。这是因为服务器端可能存在竞争条件。虽然不太可能，但RPC服务器可能会在向咱们发送答案以后，但在发送请求的确认消息以前死亡。若是发生这种状况，从新启动的RPC服务器将再次处理请求。spring-amqp客户端优雅地处理重复的响应，理想状况下RPC应该是幂等的。

 摘要

 

 

咱们的RPC将这样工做：

• Tut6Config将设置一个新的DirectExchange和一个客户端
• 客户端将利用convertSendAndReceive传递交换名称，routingKey和消息。
• 请求被发送到rpc_queue（“tut.rpc”）队列。
• RPC worker（aka：server）正在等待该队列上的请求。当请求出现时，它执行任务并使用来自replyTo字段的队列将带有结果的消息发送回客户端。
• 客户端等待回调队列上的数据。出现消息时，它会检查correlationId属性。若是它与请求中的值匹配，则返回对应用程序的响应。一样，这是经过RabbitTemplate自动完成的。

 总体来看

启动类RabbitMq0x06SpringAmqpRpcSampleApplication.java

import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Profile;
import org.springframework.scheduling.annotation.EnableScheduling;

import com.xingyun.springamqp.config.RabbitAmqpTutorialsRunner;

@SpringBootApplication
@EnableScheduling
public class RabbitMq0x06SpringAmqpRpcSampleApplication {

    public static void main(String[] args) {
        SpringApplication.run(RabbitMq0x06SpringAmqpRpcSampleApplication.class, args);
    }
    
    @Profile("usage_message")
    @Bean
    public CommandLineRunner usage() {
        return new CommandLineRunner() {

            @Override
            public void run(String... arg0) throws Exception {
                System.out.println("This app uses Spring Profiles to control its behavior.\n");
                System.out.println("Sample usage: java -jar "
                        + "RabbitMQ_0x06_SpringAMQP_RPC_Sample-0.0.1-SNAPSHOT.jar "
                        + "--spring.profiles.active=rpc"
                        + ",server");
            }
        };
    }
    
    @Profile("!usage_message")
    @Bean
    public CommandLineRunner tutorial() {
        return new RabbitAmqpTutorialsRunner();
    }
}

 

启动辅助类RabbitAmqpTutorialsRunner.java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ConfigurableApplicationContext;

public class RabbitAmqpTutorialsRunner implements CommandLineRunner {

    /**
     * application.properties文件中配置tutorial.client.duration=10000 须要
     * */
    @Value("${tutorial.client.duration:0}")
    private int duration;

    @Autowired
    private ConfigurableApplicationContext ctx;

    @Override
    public void run(String... args) throws Exception {
        // TODO Auto-generated method stub
        System.out.println("Ready ... running for " + duration + "ms");
        Thread.sleep(duration);
        ctx.close();
    }

}

 

配置类

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;

import com.xingyun.springamqp.business.Tut6Client;
import com.xingyun.springamqp.business.Tut6Server;

@Profile({ "tut6", "rpc" })
@Configuration
public class Tut6Config {
    @Profile("client")
    private static class ClientConfig {

        @Bean
        public DirectExchange exchange() {
            return new DirectExchange("tut.rpc");
        }

        @Bean
        public Tut6Client client() {
            return new Tut6Client();
        }

    }

    @Profile("server")
    private static class ServerConfig {

        @Bean
        public Queue queue() {
            return new Queue("tut.rpc.requests");
        }

        @Bean
        public DirectExchange exchange() {
            return new DirectExchange("tut.rpc");
        }

        @Bean
        public Binding binding(DirectExchange exchange, Queue queue) {
            return BindingBuilder.bind(queue).to(exchange).with("rpc");
        }

        @Bean
        public Tut6Server server() {
            return new Tut6Server();
        }

    }
}

 

Server 端

import org.springframework.amqp.rabbit.annotation.RabbitListener;

public class Tut6Server {
    
    @RabbitListener(queues = "tut.rpc.requests")
    // @SendTo("tut.rpc.replies") used when the 
    // client doesn't set replyTo.
    public int fibonacci(int n) {
        System.out.println(" [x] Received request for " + n);
        int result = fib(n);
        System.out.println(" [.] Returned " + result);
        return result;
    }

    public int fib(int n) {
        return n == 0 ? 0 : n == 1 ? 1 : (fib(n - 1) + fib(n - 2));
    }
}

 

Client 端

import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;

public class Tut6Client {
    
    @Autowired
    private RabbitTemplate template;

    @Autowired
    private DirectExchange exchange;

    int start = 0;

    @Scheduled(fixedDelay = 1000, initialDelay = 500)
    public void send() {
        System.out.println(" [x] Requesting fib(" + start + ")");
        Integer response = (Integer) template.convertSendAndReceive
            (exchange.getName(), "rpc", start++);
        System.out.println(" [.] Got '" + response + "'");
    }
}

查看用法

java -jar RabbitMQ_0x06_SpringAMQP_RPC_Sample-0.0.1-SNAPSHOT.jar

Client 端

java -jar RabbitMQ_0x06_SpringAMQP_RPC_Sample-0.0.1-SNAPSHOT.jar --spring.profiles.active=rpc,client

 

Server 端

java -jar RabbitMQ_0x06_SpringAMQP_RPC_Sample-0.0.1-SNAPSHOT.jar --spring.profiles.active=rpc,server