目录
0.交换机种类和区别
1.声明队列和交换机以及RountKey
2.初始化循环绑定
3.声明交换机
4.监听队列
4.1 监听普通队列
4.2监听死信队列
5.削峰填谷的实现
0.交换机种类和区别
Direct Exchange（直连交换机）：
直连交换机将消息发送到与消息的路由键完全匹配的队列。它是最简单的交换机类型之一。当一个队列使用某个直连交换机绑定时，它需要指定一个绑定键（binding key），当消息的路由键与该绑定键完全匹配时，消息会被发送到该队列。 Fanout Exchange（扇出交换机）：
扇出交换机会将消息发送到与其绑定的所有队列，忽略消息的路由键。当一个队列使用扇出交换机绑定时，它会接收到交换机发送的所有消息，无论消息的路由键是什么。 Topic Exchange（主题交换机）：
主题交换机根据消息的路由键和绑定键之间的模式匹配来路由消息。绑定键可以使用通配符进行匹配，支持 '*' 匹配一个单词，'#' 匹配零个或多个单词，从而允许更灵活的路由规则。 Headers Exchange（标头交换机）：
标头交换机根据消息的标头（headers）中的键值对来路由消息，而不是使用路由键。在将队列绑定到标头交换机时，可以指定一组标头键值对，只有当消息的标头中包含与绑定相匹配的所有键值对时，消息才会被路由到该队列。
如果满足key的前提下，绑定同一个交换机的队列都会分配到相同数量的信息
比如此时交换机有20条信息，a，b队列都会分配到20条信息
默认情况下，会轮询分配给消费者，也可以设置最多获取多少条未被消费的信息，根据消费者的消费能力来设置
1.声明队列和交换机以及RountKey
package com.example.config;
import lombok.Getter;
@Getter
public enum RabbitmqBind {
DATA_CLEAN_PROCESS_DEAD(
RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP,
RabbitMqQueueConstants.Q_DATA_CLEAN_PROCESS_DEAD,
RabbitmqRoutingKey.K_DATA_CLEAN_PROCESS_DEAD,
false,
false,
null,
null
),
DATA_CLEAN_PROCESS(
RabbitMqExchangeEnum.E_DIRECT_RCP,
RabbitMqQueueConstants.Q_DATA_CLEAN_PROCESS,
RabbitmqRoutingKey.K_DATA_CLEAN_PROCESS,
true,
true,
RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP,
RabbitmqRoutingKey.K_DATA_CLEAN_PROCESS_DEAD),
SMS_CLEAN_DEAD(
RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP,
RabbitMqQueueConstants.Q_API_TO_DCN_SMS_DEAD,
RabbitmqRoutingKey.K_DATA_CLEAN_FINISH_DEAD,
true,
false,
null,
null
),
SMS_CLEAN(
RabbitMqExchangeEnum.E_TOPIC_RCP,
RabbitMqQueueConstants.Q_API_TO_DCN_SMS,
RabbitmqRoutingKey.K_API_TO_DCN_SMS,
true,
true,
RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP,
RabbitmqRoutingKey.K_DATA_CLEAN_FINISH_DEAD
),
;
/**
* 交换机
*/
private RabbitMqExchangeEnum exchange;
/**
* 队列名称
*/
private String queueName;
/**
* 路由Key
*/
private RabbitmqRoutingKey routingKey;
/**
* 绑定标识
* 是否启用
*/
private Boolean isBind;
/**
* 是否绑定死信
*/
private Boolean isDeathBelief;
/**
* 绑定的死信交换机
*/
private RabbitMqExchangeEnum boundDeadExchange;
/**
* 死信key
*/
private RabbitmqRoutingKey deadRoutingKey;
RabbitmqBind(RabbitMqExchangeEnum exchange, String queueName, RabbitmqRoutingKey routingKey, Boolean isBind,
Boolean isDeathBelief, RabbitMqExchangeEnum boundDeadExchange, RabbitmqRoutingKey deadRoutingKey
) {
this.exchange = exchange;
this.queueName = queueName;
this.routingKey = routingKey;
this.isBind = isBind;
this.isDeathBelief = isDeathBelief;
this.boundDeadExchange = boundDeadExchange;
this.deadRoutingKey = deadRoutingKey;
}
/**
* 交换机
*/
@Getter
public enum RabbitMqExchangeEnum {
/**
* 交换机定义，类型 - 名称
*/
E_DIRECT_RCP("direct", "E_DIRECT_RCP"),
DEAD_E_DIRECT_RCP("direct", "DEAD_E_DIRECT_RCP"),
E_TOPIC_RCP("topic", "E_TOPIC_RCP"),
E_TOPIC_PAY("topic", "E_TOPIC_PAY");
private String exchangeType;
private String exchangeName;
RabbitMqExchangeEnum(String exchangeType, String exchangeName) {
this.exchangeType = exchangeType;
this.exchangeName = exchangeName;
}
}
/**
* 队列名定义
*/
public interface RabbitMqQueueConstants {
/**
* 接收清洗数据
*/
String Q_DATA_CLEAN_PROCESS = "RMPS_TO_RCP_DATA_CLEAN_PROCESS";
/**
* 清洗结束通知
*/
String Q_API_TO_DCN_SMS = "Q_API_TO_DCN_SMS";
/**
* 死信队列
*/
String Q_DATA_CLEAN_PROCESS_DEAD = "Q_DATA_CLEAN_PROCESS_DEAD";
/**
* 清洗结束通知死信队列
*/
String Q_API_TO_DCN_SMS_DEAD = "Q_API_TO_DCN_SMS_DEAD";
}
/**
* routingKey
*/
@Getter
public enum RabbitmqRoutingKey {
/**
* 路由
*/
K_DATA_CLEAN_PROCESS("K_DATA_CLEAN_PROCESS"),
K_API_TO_DCN_SMS("K_API_TO_DCN_SMS"),
// 路由绑定死信路由
DEAD("DEAD"),
//死信路由
K_DATA_CLEAN_PROCESS_DEAD("K_DATA_CLEAN_PROCESS_DEAD"),
K_DATA_CLEAN_FINISH_DEAD("K_DATA_CLEAN_FINISH_DEAD"),
;
private String keyName;
RabbitmqRoutingKey(String keyName) {
this.keyName = keyName;
}
}
}
2.初始化循环绑定
package com.example.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Arrays;
@Configuration
@ConditionalOnClass(EnableRabbit.class)
public class MqConfig {
@Resource
protected RabbitTemplate rabbitTemplate;
@Resource
ConnectionFactory connectionFactory;
//
// @Lazy
// @Autowired
// protected RabbitAdmin rabbitAdmin;
//
//
// public static final int DEFAULT_CONCURRENT = 10;
//
// @Bean("customContainerFactory")
// public SimpleRabbitListenerContainerFactory containerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,
// ConnectionFactory connectionFactory) {
// SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
// factory.setConcurrentConsumers(DEFAULT_CONCURRENT);
// factory.setMaxConcurrentConsumers(DEFAULT_CONCURRENT);
// configurer.configure(factory, connectionFactory);
// return factory;
// }
//
// @Bean
// @ConditionalOnMissingBean
// public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {
// return new RabbitTransactionManager(connectionFactory);
// }
//
// @Bean
// @ConditionalOnMissingBean
// public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
// return new RabbitAdmin(connectionFactory);
// }
@PostConstruct
protected void init() {
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
rabbitTemplate.setChannelTransacted(true);
//创建exchange
Arrays.stream(RabbitmqBind.RabbitMqExchangeEnum.values())
.forEach(rabbitMqExchangeEnum -> {
Exchange exchange = RabbitmqExchange
.getInstanceByType(rabbitMqExchangeEnum.getExchangeType())
.createExchange(rabbitMqExchangeEnum.getExchangeName());
rabbitAdmin.declareExchange(exchange);
}
);
//创建队列并绑定exchange
Arrays.stream(RabbitmqBind.values()).forEach(RabbitmqBind -> {
if (RabbitmqBind.getIsBind()) {
if (RabbitmqBind.getIsDeathBelief()) {
//需要绑定死信交换机的队列
rabbitAdmin.declareQueue(QueueBuilder.durable(RabbitmqBind.getQueueName())
.ttl(60000).deadLetterExchange(RabbitmqBind.getBoundDeadExchange().getExchangeName())
.deadLetterRoutingKey(RabbitmqBind.getDeadRoutingKey().getKeyName()).build());
rabbitAdmin.declareBinding(new Binding(RabbitmqBind.getQueueName(),
Binding.DestinationType.QUEUE,
RabbitmqBind.getExchange().getExchangeName(),
RabbitmqBind.getRoutingKey().getKeyName(), null));
} else {
//不需要绑定死信交换机的队列
rabbitAdmin.declareQueue(new Queue(RabbitmqBind.getQueueName(),
true, false, false, null));
rabbitAdmin.declareBinding(new Binding(RabbitmqBind.getQueueName(),
Binding.DestinationType.QUEUE,
RabbitmqBind.getExchange().getExchangeName(),
RabbitmqBind.getRoutingKey().getKeyName(), null));
}
}
});
}
}
绑定的形式由枚举类中定义
3.声明交换机
package com.example.config;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.TopicExchange;
import java.util.Arrays;
@Getter
@Slf4j
public enum RabbitmqExchange {
DIRECT("direct"){
@Override
public Exchange createExchange(String exchangeName) {
return new DirectExchange(exchangeName, true, false);
}
},
TOPIC("topic"){
@Override
public Exchange createExchange(String exchangeName) {
return new TopicExchange(exchangeName, true, false);
}
};
public static RabbitmqExchange getInstanceByType(String type){
return Arrays.stream(RabbitmqExchange.values()).filter(e -> e.getType().equals(type))
.findAny()
.orElseThrow(() ->
// new ProcessException("无效的exchange type")
new RuntimeException("无效的exchange type")
);
}
private String type;
RabbitmqExchange(String type) {
this.type = type;
}
public abstract Exchange createExchange(String exchangeName);
}
4.监听队列
4.1 监听普通队列
package com.example.listener;
import com.example.config.RabbitmqBind;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@RabbitListener(queues = {
RabbitmqBind.RabbitMqQueueConstants.Q_DATA_CLEAN_PROCESS }, concurrency = "1-5")
//, containerFactory = "customContainerFactory"
public class MqListener {
@RabbitHandler
public void processMessage(String message) {
log.info("DataClean recive message :{} ", message);
process(message);
}
@RabbitHandler
public void processMessage(byte[] message) {
String msg = new String(message);
log.info("DataClean recive message :{} ", msg);
process(msg);
}
/**
* 处理推送消息
* @param message
*/
private void process(String message) {
log.info("process message :{}" , message);
if(StringUtils.isBlank(message)) {
log.error("process message is blank , message:{}" , message);
return;
}
}
}
监听并处理任务
4.2监听死信队列
package com.example.listener;
import com.example.config.RabbitmqBind;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@RabbitListener(queues = {
RabbitmqBind.RabbitMqQueueConstants.Q_API_TO_DCN_SMS_DEAD }, concurrency = "1-5")
public class DeadListener {
@RabbitHandler
public void processMessage(String message) {
log.info("DataClean recive message :{} ", message);
process(message);
}
@RabbitHandler
public void processMessage(byte[] message) {
String msg = new String(message);
log.info("DataClean recive message :{} ", msg);
process(msg);
}
/**
* 处理推送消息
* @param message
*/
private void process(String message) {
log.info("Dead process message :{}" , message);
if(StringUtils.isBlank(message)) {
log.error("Dead process message is blank , message:{}" , message);
return;
}
}
}
5.削峰填谷的实现
把高峰期的消息填进低峰期
可以用拉取的方式来实现
或者用消费者的最大数量和最小数量来实现
channel.basicQos();//设置最大获取未确认消息的数量，实现权重
精彩内容
发表评论