RabbitMQ/RPC/TTL/死信队列

当需要调用远端计算机的函数并等待结果，这模式通常被称为远程过程调用或RPC。

BasicProperties：
消息属性 这AMQP协议预先确定了消息中的14个属性。常用的有：

  • deliveryMode
    将一个消息标记为持久化(值为2)或者瞬态的(其他值)。之前发送队列消息时的用法：

      channel.basicPublish("", "task_queue",MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

  • contentType
    用来描述媒体类型的编码。例如常常使用的JSON编码：application/json。

  • replyTo
    通常来命名回收队列的名字。

  • correlationId
    对RPC加速响应请求是很有用的。

相关性ID（Correlation Id）：
为每一个RPC请求创建一个回收队列，这个效率十分低下。每一个客户端创建一个单一的回收队列。 用Correlation Id判断队列中的响应是属于哪个请求的。
当在回收队列中接收消息时，依据这个属性值，刻意将每个响应匹配到对应的请求上。如果是未知的Correlation Id值，就丢弃这个消息，因为它不属于任何一个我们的请求。

请求RPC服务：

当客户端启动，它会创建一个匿名的独占的回收队列。 对于一个RPC请求，客户端会发送一个消息到rpc_queue队列中，其中有两个属性：replyTo（回复队列）和correlationId（每一个请求都是唯一值）。

服务端：

package testrabbitmq;
import com.rabbitmq.client.*;
import com.rabbitmq.client.AMQP.BasicProperties;

/**
 * Created by zzhblh on 2016/8/29.
 */
public class RPCServer {

    private static final String RPC_QUEUE_NAME = "rpc_queue";

    private static int fib(int n) throws Exception {
        if (n == 0) return 0;
        if (n == 1) return 1;
        return fib(n-1) + fib(n-2);
    }

    public static void main(String[] argv) throws Exception
    {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
        channel.basicQos(1);//公平分发机制
        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
        System.out.println(" [x] Awaiting RPC requests");
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            BasicProperties props = delivery.getProperties();
            BasicProperties replyProps = new BasicProperties
                    .Builder()
                    .correlationId(props.getCorrelationId())
                    .build();

            String message = new String(delivery.getBody());
            int n = Integer.parseInt(message);
            System.out.println(" [.] fib(" + message + ")");
            String response = "" + fib(n);
            channel.basicPublish( "", props.getReplyTo(), replyProps, response.getBytes());
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

客户端：

package testrabbitmq;
import com.rabbitmq.client.*;
/**
 * Created by zzhblh on 2016/8/29.
 */
public class RPCClient {
    private Connection connection;
    private Channel channel;
    private String requestQueueName = "rpc_queue";
    private String replyQueueName;
    private QueueingConsumer consumer;

    public RPCClient() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        connection = factory.newConnection();
        channel = connection.createChannel();

        replyQueueName = channel.queueDeclare().getQueue();
        consumer = new QueueingConsumer(channel);
        channel.basicConsume(replyQueueName, true, consumer);
    }

    public String call(String message) throws Exception {
        String response = null;
        String corrId = java.util.UUID.randomUUID().toString();

        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();

        channel.basicPublish("", requestQueueName, props, message.getBytes());

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response = new String(delivery.getBody());
                break;
            }
        }

        return response;
    }

    public void close() throws Exception {
        connection.close();
    }
}
  • contentType
    用来描述媒体类型的编码。例如常常使用的JSON编码，这是一个好的惯例，设置这个属性为：application/json。
  • replyTo
    通常来命名回收队列的名字。
  • correlationId
    对RPC加速响应请求是很有用的。

相关性ID（Correlation Id）：
为每一个RPC请求创建一个回收队列，这个效率十分低下。每一个客户端创建一个单一的回收队列。 用Correlation Id判断队列中的响应是属于哪个请求的。
当在回收队列中接收消息时，依据这个属性值，刻意将每个响应匹配到对应的请求上。如果是未知的Correlation Id值，就丢弃这个消息，因为它不属于任何一个我们的请求。

请求RPC服务：

当客户端启动，它会创建一个匿名的独占的回收队列。 对于一个RPC请求，客户端会发送一个消息到rpc_queue队列中，其中有两个属性：replyTo（回复队列）和correlationId（每一个请求都是唯一值）。

服务端：

package testrabbitmq;
import com.rabbitmq.client.*;
import com.rabbitmq.client.AMQP.BasicProperties;

/**
 * Created by zzhblh on 2016/8/29.
 */
public class RPCServer {

    private static final String RPC_QUEUE_NAME = "rpc_queue";

    private static int fib(int n) throws Exception {
        if (n == 0) return 0;
        if (n == 1) return 1;
        return fib(n-1) + fib(n-2);
    }

    public static void main(String[] argv) throws Exception
    {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
        channel.basicQos(1);//公平分发机制
        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
        System.out.println(" [x] Awaiting RPC requests");
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            BasicProperties props = delivery.getProperties();
            BasicProperties replyProps = new BasicProperties
                    .Builder()
                    .correlationId(props.getCorrelationId())
                    .build();

            String message = new String(delivery.getBody());
            int n = Integer.parseInt(message);
            System.out.println(" [.] fib(" + message + ")");
            String response = "" + fib(n);
            channel.basicPublish( "", props.getReplyTo(), replyProps, response.getBytes());
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

客户端：

package testrabbitmq;
import com.rabbitmq.client.*;
/**
 * Created by zzhblh on 2016/8/29.
 */
public class RPCClient {
    private Connection connection;
    private Channel channel;
    private String requestQueueName = "rpc_queue";
    private String replyQueueName;
    private QueueingConsumer consumer;

    public RPCClient() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        connection = factory.newConnection();
        channel = connection.createChannel();

        replyQueueName = channel.queueDeclare().getQueue();
        consumer = new QueueingConsumer(channel);
        channel.basicConsume(replyQueueName, true, consumer);
    }

    public String call(String message) throws Exception {
        String response = null;
        String corrId = java.util.UUID.randomUUID().toString();

        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();

        channel.basicPublish("", requestQueueName, props, message.getBytes());

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response = new String(delivery.getBody());
                break;
            }
        }

        return response;
    }

    public void close() throws Exception {
        connection.close();
    }
}

TTL

  • Queue Message TTL
    设置某队列中所有消息的TTL，通过在 queue.declare 中设置 x-message-ttl 参数，可以控制被 publish 到 queue 中的 message 被丢弃前能够存活的时间。值得注意的是，当一个 message 被路由到多个 queue 中时，他们之间不会互相影响。
Map<String, Object> args = new HashMap<String, Object>();  
args.put("x-message-ttl", 60000);//存活时间最大为 60 秒
channel.queueDeclare("myqueue", false, false, false, args);
  • Per-Message TTL
    TTL 设置可以具体到每一条 message 本身，只要在通过 basic.publish 命令发送 message 时设置 expiration 字段。expiration 字段必须为字符串类型。
AMQP.BasicProperties properties = new AMQP.BasicProperties();  
properties.setExpiration("60000");  
channel.basicPublish("myexchange", "routingkey", properties, message.getBytes());

对于第一种设置队列TTL属性的方法，一旦消息过期，就会从队列中抹去。而第二种方法里，即使消息过期，也不会马上从队列中抹去，在过期 message 到达 queue 的头部时被真正的丢弃。

  • Queue TTL
    queue 被自动删除前可以处于未使用状态的时间。未使用的意思是 queue 上没有任何 consumer ，queue 没有被重新声明，并且在过期时间段内未调用过 basic.get 命令。在服务器重启后，持久化的 queue（本来还未过期的） 的超时时间将重新计算。
Map<String, Object> args = new HashMap<String, Object>();  
args.put("x-expires", 1800000);  
channel.queueDeclare("myqueue", false, false, false, args);

DLX（死信）
利用DLX，当消息在一个队列中变成死信后，它能被重新publish到另一个Exchange，这个Exchange就是DLX。消息变成死信一向有以下几种情况：

  • 消息被拒绝（basic.reject or basic.nack）并且requeue=false
  • 消息TTL过期
  • 队列达到最大长度

DLX也是一下正常的Exchange同一般的Exchange没有区别，它能在任何的队列上被指定，实际上就是设置某个队列的属性，当这个队列中有死信时，RabbitMQ就会自动的将这个消息重新发布到设置的Exchange中去，进而被路由到另一个队列， publish可以监听这个队列中消息做相应的处理，这个特性可以弥补RabbitMQ 3.0.0以前支持的immediate参数中的向publish确认的功能。

channel.exchangeDeclare("some.exchange.name", "direct");  
Map<String, Object> args = new HashMap<String, Object>();  
args.put("x-dead-letter-exchange", "some.exchange.name");  
channel.queueDeclare("myqueue", false, false, false, args);

你也可以为这个DLX指定routing key，如果没有特殊指定，则使用原队列的routing key

args.put("x-dead-letter-routing-key", "some-routing-key");

http://memorynotfound.com/produce-consume-rabbitmq-spring-json-message-queue/
http://blog.csdn.net/jiao_fuyou/article/details/22923935

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 221,576评论 6赞 515
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 94,515评论 3赞 399
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 168,017评论 0赞 360
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 59,626评论 1赞 296
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 68,625评论 6赞 397
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 52,255评论 1赞 308
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 40,825评论 3赞 421
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 39,729评论 0赞 276
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 46,271评论 1赞 320
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 38,363评论 3赞 340
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 40,498评论 1赞 352
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 36,183评论 5赞 350
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 41,867评论 3赞 333
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 32,338评论 0赞 24
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 33,458评论 1赞 272
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 48,906评论 3赞 376
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 45,507评论 2赞 359

推荐阅读更多精彩内容

  • Spring Cloud
    Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智...
    卡卡罗2017阅读 134,699评论 18赞 139
  • RabbitMQ
    来源 RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。支持消息的持久化、事务、拥塞控...
    jiangmo阅读 10,367评论 2赞 34
  • 浅析消息队列之rabbitMQ
    rabbitMQ是一款基于AMQP协议的消息中间件，它能够在应用之间提供可靠的消息传输。在易用性，扩展性，高可用性...
    点融黑帮阅读 3,007评论 3赞 41
  • 消息队列探秘-RabbitMQ消息队列介绍
    1. 历史 RabbitMQ是一个由erlang开发的AMQP（Advanced Message Queue ）的...
    高广超阅读 6,096评论 3赞 51
  • 消息队列及rabbitmq简介
    1.什么是消息队列 消息队列允许应用间通过消息的发送与接收的方式进行通信，当消息接收方服务忙或不可用时，其提供了一...
    zhuke阅读 4,476评论 0赞 12