1、安装rabbitmq
docker安装,拉取镜像
docker pull rabbitmq:management
创建容器并启动
docker run -d --name rabbitmq --publish 5671:5671 --publish 5672:5672 --publish 4369:4369 --publish 25672:25672 --publish 15671:15671 --publish 15672:15672 rabbitmq:management
管理地址:
http://localhost:15672/ 用户名:guest 密码:guest
2、配置springboot
(1)pom.xml添加依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
(2)添加rabbit配置文件,配置server信息
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
- @Author 冯战魁
- @Date 2018/1/12 下午2:50br/>*/
@Configuration
public class AmqpConfig {br/>@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setAddresses("127.0.0.1:5672");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");
connectionFactory.setPublisherConfirms(true); //必须要设置
return connectionFactory;br/>}
@Bean
public Queue helloQueue() {
return new Queue("hello");
}
}
@RestController
public class RabbitSenderController {br/>@Autowired
private AmqpTemplate rabbitTemplate; br/>@RequestMapping("/sender")
public void sender(){
String[] tasks = {"1.","2..","3...","4...."};
for (int i=0;i<tasks.length;i++){
String content = tasks[i];
System.out.println("Sender : " + content);
this.rabbitTemplate.convertAndSend("hello",content);
}
}
}
(4)创建两个消费者Receiver1,Receiver2处理队列的任务
@Author 冯战魁 @Date 2018/1/12 下午3:06br/>*/@Component
@RabbitListener(queues = "hello")
public class RabbitReceiver1 {br/>@RabbitHandler
public void process(String content){
System.out.println("Receiver1 : " + content);
try {
dowork(content);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void dowork(String content) throws InterruptedException{
for (char ch: content.toCharArray()) {
if (ch == '.'){
Thread.sleep(1000);
}
}
}
} @Author 冯战魁 @Date 2018/1/12 下午3:06br/>*/
@Component
@RabbitListener(queues = "hello")
public class RabbitReceiver2 {br/>@RabbitHandler
public void process(String content){
System.out.println("Receiver2 : " + content);
try {
dowork(content);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void dowork(String content) throws InterruptedException{
for (char ch: content.toCharArray()) {
if (ch == '.'){
Thread.sleep(1000);
}
}
}
}
(5)验证代码
启动springboot,执行生产者的接口
http://localhost:8080/sender
然后查看终端输出的信息,如图所示
可以看到最先完成的消费者去执行下一个任务,是按任务时长执行,并不是按任务个数平分,防止消费者无意义的等待,提高效率。
查看rabbit的管理端http://localhost:15672/
可以看到已经生成队列
至此整个任务完成!