RabbitMQ 消息属性答案

【问题标题】：RabbitMQ Message PropertiesRabbitMQ 消息属性
【发布时间】：2012-05-29 19:57:55
【问题描述】：

我在 RabbitMQ 中有这样的代码：

byte[] rawBytes = serialize(trxEntities);    
byte[] zipped = rawBytes;  
if (shouldBeCompress) {  
zipped = compressor.compress(rawBytes);  
}  
BasicProperties persistentBasic = MessageProperties.PERSISTENT_BASIC;  
channel.basicPublish("", queueName, persistentBasic, zipped);

正如您所见，我的一些消息应该与其他一些不应该一起压缩。
有什么方法可以设置任何属性来告诉消费者“嘿！这是一条压缩消息”？

PS。 “com.rabbitmq.client.AMQP.BasicProperties.BasicProperties(..., Map headers, ...)”对我有帮助吗？我的意思是我可以在 BasicProperties.header 中设置任何参数吗？

【问题讨论】：

标签： java rabbitmq

【解决方案1】：

我认为您可以在标题中添加任何您喜欢的内容。但是，有一个名为“contentEncoding”的字段，我认为它更适合这种情况。您可以在此字段中输入“gzip”、“deflate”或压缩算法，查看此页面以了解为 HTTP 定义的编码：http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.5

【讨论】：

【解决方案2】：

你可以为它使用一个包装器：

    public class wrapper(){
    public boolean isZipped;
    public String serializedMessage;
}

然后序列化这条消息 Java Serializable Object to Byte Array

或者你可以使用这个代码：

persistentBasic = persistentBasic.builder().headers(filter).build();

并将适当的过滤器放在标题中。

【讨论】：

【解决方案3】：

对于我正在使用的使用 RabbitMQ 的项目，我们使用标头字段来标识内容类型和内容编码

例如，队列中的纯文本消息会读取

{'type' : 'plaintext', 'encoding' : 'utf-8'}

压缩后的数据流经过Base64处理后发送

{'type' : 'gzip', 'encoding' : 'base64'}

RabbitMQ 没有标准化的机制来识别内容类型和编码，您可以采用自己的标准或选择一个常用的标准。

【讨论】：

不应该键入和编码类似于它的使用更广泛吗？即stackoverflow.com/a/23600787/98069

【解决方案4】：

是的，你可以！

我们使用type 字段指定发送的消息（例如，USER_INFO、HEARTBEAT 等），使用contentEncoding 字段指定压缩（如果有）：

AMQP.BasicProperties.Builder propsBuilder = new AMQP.BasicProperties.Builder();
propsBuilder.type(typeName);
if (compress)
{
  propsBuilder.contentEncoding("zip");
}
BasicProperties props = propsBuilder.build();
channel.basicPublish(targetExchange, "", true, props, data);

// and receiving works like this:
Delivery delivery = consumer.nextDelivery();
byte[] data = delivery.getBody();
BasicProperties props = delivery.getProperties();
String typeName = props.getType();
String replyToServerId = props.getReplyTo();
String contentEncoding = props.getContentEncoding();

【讨论】：