环境
- jdk1.8
- rabbitmq 3.7.9
1 创建项目
-
idea创建项目时可以直接引入rabbitmq的依赖。
image.png
生成的pom:
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.4.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>springboot_rabbitmq</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>springboot_rabbitmq</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
spring-boot-starter-amqp 的版本会自动使用与spring-boot-starter-parent版本匹配的版本。
2 配置
application.properties:
spring.application.name=springboot-rabbitmq
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.virtual-host=/
或者 application.yml:
spring:
application:
name: springboot-rabbitmq
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
publisher-confirms: true
virtual-host: /
配置了broker的ip、端口、用户名、密码、vhost,这里的 publisher-confirms: true
表示在exchange成功发送消息到queue以后,会给生产者发送确认消息。这一部分下一章详细讲解。
3 初始化
package com.example.springboot_rabbitmq.rabbitmq;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author: zenghong
* @Date: 2019/4/25 09:42
* @Description: TODO
*/
@Configuration
public class RabbitMQInit {
final static String queueName = "hello";
@Bean
public Queue helloQueue() {
return new Queue("hello");
}
}
这里初始化了一个名为 hello
的queue
4 编写生产者和消费者
package com.example.springboot_rabbitmq.rabbitmq.sender;
import org.springframework.amqp.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Date;
/**
* @author: zenghong
* @Date: 2019/4/25 09:45
* @Description: TODO
*/
@Component
public class HelloSender {
@Autowired
private RabbitTemplate rabbitTemplate;
public void send() {
String sendMsg = "hello1 " + new Date();
System.out.println("Sender1 : " + sendMsg);
this.rabbitTemplate.convertAndSend("hello", sendMsg);
}
}
this.rabbitTemplate.convertAndSend 第一个参数是 routing key,第二个参数是发送的msg。
package com.example.springboot_rabbitmq.rabbitmq.consumer;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @author: zenghong
* @Date: 2019/4/25 09:46
* @Description: TODO
*/
@Component
@RabbitListener(queues = "hello")
public class HelloCosumer {
@RabbitHandler
public void process(String hello) {
System.out.println("Receiver1 : " + hello);
}
}
消费者监听名为 hello
的queue
5 启动
Connected to the target VM, address: '127.0.0.1:49579', transport: 'socket'
. ____ _ __ _ _
/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
\\/ ___)| |_)| | | | | || (_| | ) ) ) )
' |____| .__|_| |_|_| |_\__, | / / / /
=========|_|==============|___/=/_/_/_/
:: Spring Boot :: (v2.1.4.RELEASE)
2019-04-25 10:33:48.460 INFO 192 --- [ main] c.e.s.SpringbootRabbitmqApplication : Starting SpringbootRabbitmqApplication on cenghongdeMacBook-Pro.local with PID 192 (/Users/zenghong/Documents/code/other/wed-job/springboot_rabbitmq/target/classes started by zenghong in /Users/zenghong/Documents/code/other/wed-job/springboot_rabbitmq)
2019-04-25 10:33:48.463 INFO 192 --- [ main] c.e.s.SpringbootRabbitmqApplication : No active profile set, falling back to default profiles: default
2019-04-25 10:33:49.591 INFO 192 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat initialized with port(s): 8080 (http)
2019-04-25 10:33:49.613 INFO 192 --- [ main] o.apache.catalina.core.StandardService : Starting service [Tomcat]
2019-04-25 10:33:49.613 INFO 192 --- [ main] org.apache.catalina.core.StandardEngine : Starting Servlet engine: [Apache Tomcat/9.0.17]
2019-04-25 10:33:49.713 INFO 192 --- [ main] o.a.c.c.C.[Tomcat].[localhost].[/] : Initializing Spring embedded WebApplicationContext
2019-04-25 10:33:49.713 INFO 192 --- [ main] o.s.web.context.ContextLoader : Root WebApplicationContext: initialization completed in 1204 ms
2019-04-25 10:33:50.073 INFO 192 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 'applicationTaskExecutor'
2019-04-25 10:33:50.431 INFO 192 --- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [127.0.0.1:5672]
2019-04-25 10:33:50.467 INFO 192 --- [ main] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory#396ef8b2:0/SimpleConnection@6949e948 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 49587]
2019-04-25 10:33:50.571 INFO 192 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat started on port(s): 8080 (http) with context path ''
2019-04-25 10:33:50.576 INFO 192 --- [ main] c.e.s.SpringbootRabbitmqApplication : Started SpringbootRabbitmqApplication in 2.708 seconds (JVM running for 4.35)
启动成功以后可以看到控制台已经创建好了hello
队列:
并且bind了默认的exchange:
image.png
我们看一看默认的exchange:
image.png
可以看出,它是一个direct的exchange。 不是说好的direct需要routing key和binding key完全匹配才能接收消息么? 注意这句话:
image.png
意思是说,这个默认的exchange 会绑定到每一个queue, 并且会把消息发送给 队列名与routing key完全相同的队列(类似于以queue name为binding key)。 所以上面咱们的sender的代码写的routing key 是 hello,与我们创建的queue名字一样。
6 测试
6.1 编写测试类
package com.example.springboot_rabbitmq.controller;
import com.example.springboot_rabbitmq.rabbitmq.sender.HelloSender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @author: zenghong
* @Date: 2019/4/25 09:49
* @Description: TODO
*/
@RestController
@RequestMapping("/rabbit")
public class RabbitMqController {
@Autowired
private HelloSender helloSender1;
@PostMapping("/hello")
public String hello() {
helloSender1.send();
return "success";
}
}
这里写了一个rest接口来触发消息发送。
6.2 发送请求 到 http://localhost:8080/rabbit/hello
image.png
后台日志:
从后台日志可以看到消息一经成功发送和被接收了。
再看看控制台:
image.png
我们可以看到,有一条消息被发送,并且被接收了。
网友评论