添加依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
添加配置
- 配置文件中新增配置
#rabbitmq
spring.rabbitmq.host=10.110.3.62
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
#消费者数量
spring.rabbitmq.listener.simple.concurrency= 10
spring.rabbitmq.listener.simple.max-concurrency= 10
#消费者每次从队列获取的消息数量
spring.rabbitmq.listener.simple.prefetch= 1
#消费者自动启动
spring.rabbitmq.listener.simple.auto-startup=true
#消费失败，自动重新入队
spring.rabbitmq.listener.simple.default-requeue-rejected= true
#启用发送重试
spring.rabbitmq.template.retry.enabled=true
spring.rabbitmq.template.retry.initial-interval=1000
spring.rabbitmq.template.retry.max-attempts=3
spring.rabbitmq.template.retry.max-interval=10000
spring.rabbitmq.template.retry.multiplier=1.0
- 新建发送者
@Service
public class MQSender {
private static Logger log = LoggerFactory.getLogger(MQSender.class);
@Autowired
AmqpTemplate amqpTemplate ;
public void sendMiaoshaMessage(MiaoshaMessage mm) {
String msg = RedisService.beanToString(mm);
log.info("send message:"+msg);
amqpTemplate.convertAndSend(MQConfig.TEST_QUEUE, msg);
}
}
- 新建接收者
@Service
public class MQReceiver {
private static Logger log = LoggerFactory.getLogger(MQReceiver.class);
@RabbitListener(queues=MQConfig.TEST_QUEUE)
public void receive(String message) {
log.info("receive message:"+message);
//需要进行的操作
}
}
- 新建配置类，用来配置我们需要的bean。
@Configuration
public class MQConfig {
public static final String TEST_QUEUE = "test.queue";
public static final String QUEUE = "queue";
/**
* Direct模式 交换机Exchange
* */
@Bean
public Queue testQueue() {
return new Queue(TEST_QUEUE, true);
}
/**
* Direct模式 交换机Exchange
* */
@Bean
public Queue queue() {
return new Queue(QUEUE, true);
}
}