本文主要是介绍消息队列---RabbitMQ的学习(七)，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

 做想去做的事情，不要在乎结局

Success

• 自述
• RPC
• RPC场景模型解析
• 消息属性
• CorrelationId
• 回调队列
• 疑难问题
• Ending

自述

    RabbitMQ的第六个场景------远程过程调用(RPC)，该文章的内容也不能说较多吧，但比起之前的文章，该文章是有些难度的，实际上最主要的就是将之前所学的结合了起来，不过不用担心，我会通过注释的方式进行介绍。

RPC

   RPC，各位应该非常熟悉了吧，它是计算中非常常见的模式，一种设计思想，而该场景则就是使用RabbitMQ构建一个RPC系统，一个客户端和一个RPC服务端。

RPC场景模型解析

    RabbitMQ进行RPC很容易，客户端发送请求，服务端回复响应消息，为了接收服务端响应消息，我们需要随着请求一起发送“回调“队列地址，核心流程为：客户端发送请求到“请求”队列中，服务端从“请求”队列中获取客户端请求消息并进行处理，处理完成之后将响应消息发送到“回调”队列中，最终客户端在从“回调”队列中获取消息。
     ①客户端发送请求到“请求”队列中。
     ②RPC服务端实时接收“请求”队列，每当“请求”队列中有新的请求消息时，会立刻进行接收并做出处理。
     ③RPC服务端每处理完成一个请求时，都会将响应信息发送到“回调”队列中。
     ④客户端等待“回调”队列，当“回调”队列中有消息时将获取该消息，并与该消息的correlationid属性进行对比，如果匹配上则代表该响应消息是对应我们请求的响应消息，如果匹配不上，则丢弃。

消息属性

BasicProperties常定义的几种属性：
deliverMode:将消息标记为持久。
contentType：接收内容类型，例如：application/json。
replyto：回调队列的名称。
correlationid：用于将RPC响应与请求相关联。

CorrelationId

   如果单看它的话可能不太理解，但是如果要是换个称呼“服务寻址”或“寻址”，各位大佬是不是瞬间就想到了对应的场景，至于我们为什么需要correlationid呢？举个例子：就拿存储箱举例吧，每次我们到超市购物的时候，都会发现超市门口旁边会有一个存储箱，而当我们将东西存储到存储箱里面的时候，存储箱会打印出一个条形码，其作用就是用于标识，标识你的物品由该条形码代替，当你想取走你存储的东西时，将该条形码进行匹配，如果存储箱能匹配到该条形码的话，则代表该存储箱的确存储着你的东西，如果匹配不到，咦~，而在实际业务中，我们可能会接收到多条响应消息，但并不代表多条响应消息都是针对一个请求返回的，所以我们将correlationid属性进行匹配。
       注: 不是所有的服务寻址都采用此方式。
                    

回调队列

   RPC服务端
 RpcServer.java

/**
 * rpc服务端 ==rpc工作者
 * 接收请求并进行处理请求，将处理好的请求传递给replyto队列
 */
public class RpcServer {
    //queuename
    private static final String RPC_QUEUE = "rpq_queue";

    /**
     * 斐波那契数列
     */
    private static int fib(int n) {
        if (n == 0) return 0;
        if (n == 1) return 1;
        return fib(n - 1) + fib(n - 2);
    }

    public static void main(String[] args) {
        System.out.println("=========服务端启动成功=========");
        try {
            //1.connection
            Connection connection = ConnectionUtil.getConnection();
            //2.channel
            Channel channel = connection.createChannel();
            //3.queue
            channel.queueDeclare(RPC_QUEUE, false, false, false, null);
            //4.clear --每次都清楚队列中的消息或该队列存储在内存、磁盘上的消息
            channel.queuePurge(RPC_QUEUE);
            //5.公平调度 --防止存在服务器x压力而服务器x2压力小的情况
            channel.basicQos(1);
            //6.临时对象  --同步块标识
            Object monitor = new Object();
            //7.回调方法 --内部包含传递响应消息与客户端
            DeliverCallback deliverCallback = new DeliverCallback() {
                @Override
                public void handle(String s, Delivery delivery) throws IOException {
                    AMQP.BasicProperties replyProps = new AMQP.BasicProperties
                            .Builder()
                            .correlationId(delivery.getProperties().getCorrelationId())
                            .build();
                    //响应消息
                    String response = "";
                    try {
                        //接收客户端发送的请求
                        String message = new String(delivery.getBody());
                        //将客户端发送的消息转换
                        int n = Integer.parseInt(message);
                        //返回斐波那契数
                        response += fib(n);
                    } catch (Exception e) {
                        System.out.println(e.getMessage());
                    } finally {
                        //传递消息---到客户端
                        channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes());
                        //手动ack响应 --告知传递着消息接收成功
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                        //同步块
                        synchronized (monitor) {
                            //线程放行
                            monitor.notify();
                        }
                    }
                }
            };
            //8.在指定的队列中获取消息
            //这里的false代表的是手动调用ack消息应答
            channel.basicConsume(RPC_QUEUE, false, deliverCallback, cancel -> {
            });
            //9.RPC服务器挂起状态，等待客户端请求信息
            while (true) {
                synchronized (monitor) {
                    try {
                        //线程等待
                        monitor.wait();
                        System.out.println("成功处理客户端请求-------");
                    } catch (Exception e) {
                        System.out.println(e.getMessage());
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

   RPC客户端
 RpcClient.java

/**
 * Rpc客户端
 * 请求消息于Rpc服务端，接收Rpc服务端的响应信息(通过correlationid确定指定的响应信息)
 */
public class RpcClient implements AutoCloseable {

    //connection
    private Connection connection;
    //channel
    private Channel channel;
    //queuename
    private static final String RPC_QUEUE = "rpq_queue";

    //构造器
    RpcClient() throws Exception {
        connection = ConnectionUtil.getConnection();
        channel = connection.createChannel();
    }

    public static void main(String[] args) {
        try (RpcClient rpcClient = new RpcClient()) {
            //模拟请求
            for (int i = 0; i < 5; i++) {
                String i_str = Integer.toString(i);
                String response = rpcClient.call(i_str);
                System.out.println("服务端返回的消息========" + response);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送信息与request队列中----模拟请求
     *
     * @param msg
     * @return
     */
    public String call(String msg) throws IOException, InterruptedException {
        //标识id --为于rpc服务端返回的消息确认
        final String corrid = UUID.randomUUID().toString();
        //随机队列 --这里队列名称不重要，只是为了接收服务端响应消息
        String queue_random = channel.queueDeclare().getQueue();
        AMQP.BasicProperties basicProperties = new AMQP.BasicProperties
                .Builder()
                .replyTo(queue_random)
                .correlationId(corrid)
                .build();
        //将消息发送到指定的队列中
        channel.basicPublish("", RPC_QUEUE, basicProperties, msg.getBytes());
        System.out.println("==消息发送成功==");
        //阻塞队列 --juc(java.util.concurrent)包下
        final BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(1);
        //回调
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                //判断 --目录:将指定corrid标识的消息进行返回
                if (corrid.equals(delivery.getProperties().getCorrelationId())) {
                    //存储信息 --RPC服务端返回的响应信息
                    blockingQueue.offer(new String(delivery.getBody()));
                }
            }
        };
        //消费指定的队列
        String ctag = channel.basicConsume(queue_random, deliverCallback, cancel -> {
        });
        //获取并删除队列的头
        String result = blockingQueue.take();
        //删除
        channel.basicCancel(ctag);
        return result;
    }

    /**
     * 释放资源
     *
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
        connection.close();
        System.out.println("=======最后执行=======");
    }
}

疑难问题

   1.如果没有RPC服务器在运行，客户端应该如何反应？
        如果RPC服务器没有在运行，不排除服务器已经挂掉、宕机、网络崩溃等多种情况，为此解决方案根据实际业务来定制，例如：搭建RPC集群、定期监控作业等。
   2.客户端是否应该为 RPC 设置某种超时？
        根据实际业务场景进行设置，默认推荐设置RPC超时，也算是属于一个小小的优化。
   3.如果服务器出现故障并引发异常，是否应该将其转发给客户端？
        个人想法：可以将其转发给客户端，当客户端匹配到该特殊消息时，则做出相对应的友好处理。

Ending

      各位，端午节安康，记得吃粽子哦。

这篇关于消息队列---RabbitMQ的学习(七)的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！