问题
在用Spring boot 的 spring-boot-starter-amqp 快速启动 rabbitMq 是遇到了个坑
消费者端获取不到：correlationId或CorrelationIdString
问题产生的原因
correlationId 的在 spring rabbitmq 2.0 以后 byte方式会被放弃，所以 目前 代码中有些地方没有改过来，应该算一个BUG
@SuppressWarnings("deprecation")
public class DefaultMessagePropertiesConverter implements MessagePropertiesConverter {
@Deprecated
public enum CorrelationIdPolicy {
STRING, BYTES, BOTH
}
private static final int DEFAULT_LONG_STRING_LIMIT = 1024;
private final int longStringLimit;
private final boolean convertLongLongStrings;
private volatile CorrelationIdPolicy correlationIdPolicy = CorrelationIdPolicy.BYTES;
}
/**
* For inbound, determine whether correlationId, correlationIdString or
* both are populated. For outbound, determine whether correlationIdString
* or correlationId is used when mapping; if {@code CorrelationIdPolicy.BOTH}
* is set for outbound, String takes priority and we fallback to bytes.
* Default {@code CorrelationIdPolicy.BYTES}.
* @param correlationIPolicy true to use.
* @deprecated - the byte[] version of correlation id will be removed in 2.0
*/
@Deprecated
public void setCorrelationIdPolicy(CorrelationIdPolicy correlationIPolicy) {
setCorrelationIdAsString(correlationIPolicy);
} @SuppressWarnings("deprecation")
public MessageProperties toMessageProperties(final BasicProperties source, final Envelope envelope,
final String charset) {
MessageProperties target = new MessageProperties();
Map<String, Object> headers = source.getHeaders();
if (!CollectionUtils.isEmpty(headers)) {
for (Map.Entry<String, Object> entry : headers.entrySet()) {
String key = entry.getKey();
if (MessageProperties.X_DELAY.equals(key)) {
Object value = entry.getValue();
if (value instanceof Integer) {
target.setReceivedDelay((Integer) value);
}
}
else {
target.setHeader(key, convertLongStringIfNecessary(entry.getValue(), charset));
}
}
}
target.setTimestamp(source.getTimestamp());
target.setMessageId(source.getMessageId());
target.setReceivedUserId(source.getUserId());
target.setAppId(source.getAppId());
target.setClusterId(source.getClusterId());
target.setType(source.getType());
Integer deliveryMode = source.getDeliveryMode();
if (deliveryMode != null) {
target.setReceivedDeliveryMode(MessageDeliveryMode.fromInt(deliveryMode));
}
target.setDeliveryMode(null);
target.setExpiration(source.getExpiration());
target.setPriority(source.getPriority());
target.setContentType(source.getContentType());
target.setContentEncoding(source.getContentEncoding());
String correlationId = source.getCorrelationId();
if (!CorrelationIdPolicy.BYTES.equals(this.correlationIdPolicy) && correlationId != null) {
target.setCorrelationIdString(correlationId);
}
if (!CorrelationIdPolicy.STRING.equals(this.correlationIdPolicy)) {
if (correlationId != null) {
try {
target.setCorrelationId(source.getCorrelationId().getBytes(charset));
}
catch (UnsupportedEncodingException ex) {
throw new AmqpUnsupportedEncodingException(ex);
}
}
}
String replyTo = source.getReplyTo();
if (replyTo != null) {
target.setReplyTo(replyTo);
}
if (envelope != null) {
target.setReceivedExchange(envelope.getExchange());
target.setReceivedRoutingKey(envelope.getRoutingKey());
target.setRedelivered(envelope.isRedeliver());
target.setDeliveryTag(envelope.getDeliveryTag());
}
return target;
} public BasicProperties fromMessageProperties(final MessageProperties source, final String charset) {
BasicProperties.Builder target = new BasicProperties.Builder();
target.headers(this.convertHeadersIfNecessary(source.getHeaders()))
.timestamp(source.getTimestamp())
.messageId(source.getMessageId())
.userId(source.getUserId())
.appId(source.getAppId())
.clusterId(source.getClusterId())
.type(source.getType());
MessageDeliveryMode deliveryMode = source.getDeliveryMode();
if (deliveryMode != null) {
target.deliveryMode(MessageDeliveryMode.toInt(deliveryMode));
}
target.expiration(source.getExpiration())
.priority(source.getPriority())
.contentType(source.getContentType())
.contentEncoding(source.getContentEncoding());
@SuppressWarnings("deprecation")
byte[] correlationId = source.getCorrelationId();
String correlationIdString = source.getCorrelationIdString();
if (!CorrelationIdPolicy.BYTES.equals(this.correlationIdPolicy)
&& StringUtils.hasText(correlationIdString)) {
target.correlationId(correlationIdString);
correlationId = null;
}
if (!CorrelationIdPolicy.STRING.equals(this.correlationIdPolicy)
&& correlationId != null && correlationId.length > 0) {
try {
target.correlationId(new String(correlationId, charset));
}
catch (UnsupportedEncodingException ex) {
throw new AmqpUnsupportedEncodingException(ex);
}
}
String replyTo = source.getReplyTo();
if (replyTo != null) {
target.replyTo(replyTo);
}
return target.build();
}
解决方法
生产者：
public void topicPublish(String content) {
CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend(AmqpDirectExchangeConfig.FANOUT_EXCHANGE,"",
this.buildMessage(content,correlationId.getId()), correlationId);
this.log.info("消息id-{}，消息内容为-{}，已发送",correlationId,content);
}
/**
* 返回rabbitTemplate
* @return
*/
@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
//必须是prototype类型
public RabbitTemplate rabbitRtryTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionAckFactory());
template.setMessagePropertiesConverter(defaultMessagePropertiesConverter());
template.setRetryTemplate(rabbitRetry());
template.setReplyTimeout(2000);//2s秒超时
return template;
}
@Bean
public MessagePropertiesConverter defaultMessagePropertiesConverter(){
DefaultMessagePropertiesConverter messagePropertiesConverter=new DefaultMessagePropertiesConverter();
messagePropertiesConverter.setCorrelationIdPolicy(DefaultMessagePropertiesConverter.CorrelationIdPolicy.STRING);
return messagePropertiesConverter;
}
消费者：
/**
* 消息消费者
* @return
*/
@Bean
public SimpleMessageListenerContainer messageContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionAckFactory());
container.setQueues(queue1());
container.setExposeListenerChannel(true);
container.setMessagePropertiesConverter(defaultMessagePropertiesConverter());
container.setMaxConcurrentConsumers(1);
container.setConcurrentConsumers(1);
container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
container.setMessageListener(new ChannelAwareMessageListener () {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
byte[] body = message.getBody();
MessageProperties messageProperties=message.getMessageProperties();
log.info("消费者A,从队列{}，订阅到CorrelationId=[{}],消息body=[{}]",
messageProperties.getConsumerQueue(),
messageProperties.getCorrelationIdString(),
new String(body,"utf-8"));
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费
}
});
return container;
}