消息队列整合SpringBoot和RabbitMQ

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.mq-amqp</groupId>
<artifactId>mq-amqp-client</artifactId>
<version>1.0.5</version>
</dependency>

@Component
public class Producer {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 发生消息到RabbitMQ,使用SpringBoot默认的交换机
*
* @param message
*/
public void sendMessage(String message) {
rabbitTemplate.setConfirmCallback(confirmCallback);
rabbitTemplate.convertAndSend(RabbitMQConfig.SIMPLE_QUEUE_NAME, message);
}
//回调函数: confirm确认 此处的confirm确认是发送的ACK确认
final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if(ack){
//如果confirm返回成功 则进行更新
System.out.println("confirm确认消息已经发出");
} else {
//失败则进行具体的后续操作:重试 或者补偿等手段
System.err.println("异常处理...");
}
}
};
}

    @Autowired
private RabbitTemplate rabbitTemplate;
//记住这个，这Template是SpringBoot里面整合RabbitMQ自带的，不是我自己专门写的哈，大家不要一看到这个注入，就寻思是个自己写的。
/**
* 普通默认点对点的消息消费者
* @param message
*/
@RabbitListener(queues = {RabbitMQConfig.SIMPLE_QUEUE_NAME})
@RabbitHandler
public void receiveMessage(String message, Channel channel, @Headers Map<String,Object> headers) { //
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//处理监听得到的消息
String message = null;
try {
message = new String(body, "UTF-8");
//消息处理逻辑
sendMessage(message);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
channel.abort();
} finally {
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
//消息消费完成确认
try {
channel.basicConsume(RabbitMQConfig.SIMPLE_QUEUE_NAME, autoAck, consumer);
}catch (Exception e){
}
System.out.println("消费者进行消息消费：消费主体是-----"+ message);
}

        //消息发送
for (int i = 0;i < 100 ;i++){
producer.sendMessage("生产消息消息"+i);
}
消费者一模一样的复制出一份，在消费消息的时候打印一下就可以啦

/**
* Fanout型交换机MQ模型:订阅模式，消息到达交换机会转发到与该交换机绑定的队列
*/
/**
* 与fanout绑定的第一个队列
*/
public static final String FIRST_FANOUT_QUEUE_NAME = "First_Fanout_Queue";
/**
* 与fanout交换机绑定的第二个队列
*/
public static final String SECOND_FANOUT_QUEUE_NAME = "Send_Fanout_Queue";
/**
* fanout 交换机
*/
public static final String FANOUT_EXCHANGE_NAME = "Fanout_Exchange";
/**
* FanoutExchange,持久化、非自动删除
*
* @return
*/
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange(FANOUT_EXCHANGE_NAME);
}
@Bean
public Queue firstFanoutQueue() {
return new Queue(FIRST_FANOUT_QUEUE_NAME);
}
@Bean
public Queue secondFanoutQueue() {
return new Queue(SECOND_FANOUT_QUEUE_NAME);
}
@Bean
public Binding firstFanoutBinding() {
return BindingBuilder.bind(firstFanoutQueue()).to(fanoutExchange());
}
@Bean
public Binding secondFanoutBinding() {
return BindingBuilder.bind(secondFanoutQueue()).to(fanoutExchange());
}

  /**
* Fanout型交换机的MQ模型消费消息
* @param message
*/
@RabbitListener(queues = {RabbitMQConfig.FIRST_FANOUT_QUEUE_NAME})
@RabbitHandler
public void receiveFanoutMessage(String message) {
System.out.println("11111Fanout消费者进行消息消费：消费主体是-----"+ message);
}
/**
* Fanout型交换机的MQ模型消费消息
* @param message
*/
@RabbitListener(queues = {RabbitMQConfig.SECOND_FANOUT_QUEUE_NAME})
@RabbitHandler
public void receiveFanoutMessage2(String message) {
System.out.println("2222Fanout消费者进行消息消费：消费主体是-----"+ message);
}

/**
* 发送消息至fanout交换机,由于fanout只关注订阅关系，所以routing key随便指定都可以
* @param message
*/
public void sendFanoutMessage(String message){
rabbitTemplate.convertAndSend(RabbitMQConfig.FANOUT_EXCHANGE_NAME, "------", message);
}