赞
踩
目录
2.1 导入spring-boot-starter-amqp依赖
普通项目导入依赖
<!--指定 jdk 编译版本--> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <configuration> <source>8</source> <target>8</target> </configuration> </plugin> </plugins> </build> <dependencies> <!--rabbitmq 依赖客户端--> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.8.0</version> </dependency> <!--操作文件流的一个依赖--> <dependency> <groupId>commons-io</groupId> <artifactId>commons-io</artifactId> <version>2.6</version> </dependency> </dependencies>
springboot工程导入依赖
<!--SpringBoot整合AMQP-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
- public class Producer {
- public static final String QUEUE_NAME = "queue1";
- public static void main(String[] args) {
- // 建立连接工厂,但是没连接
- ConnectionFactory connectionFactory = new ConnectionFactory();
- connectionFactory.setHost("192.168.35.132");
- connectionFactory.setUsername("itcast");
- connectionFactory.setPassword("123321");
- try {
- // 创建一个连接
- Connection connection = connectionFactory.newConnection();
- // 创建一个信道
- Channel channel = connection.createChannel();
- // 创建一个队列
- /**
- * 生成一个队列
- * 1.队列名称
- * 2.队列里面的消息是否持久化 默认消息存储在内存中
- * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
- * 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除
- * 5.其他参数
- */
- channel.queueDeclare(QUEUE_NAME,false,false,false,null);
- // 发送消息
- /**
- * 发送一个消息
- * 1.发送到那个交换机
- * 2.路由的 key 是哪个
- * 3.其他的参数信息
- * 4.发送消息的消息体
- */
- channel.basicPublish("",QUEUE_NAME,null,"你好".getBytes());
- System.out.println("消息发送成功");
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- }
- public class Consumer {
-
- public static final String QUEUE_NAME = "queue1";
-
- public static void main(String[] args) {
- // 建立连接工厂,但是没连接
- ConnectionFactory connectionFactory = new ConnectionFactory();
- connectionFactory.setHost("192.168.35.132");
- connectionFactory.setUsername("itcast");
- connectionFactory.setPassword("123321");
- try {
- Connection connection = connectionFactory.newConnection();
- Channel channel = connection.createChannel();
- //推送的消息如何进行消费的接口回调
- DeliverCallback deliverCallback=(consumerTag,delivery)-> {
- String message= new String(delivery.getBody());
- System.out.println(message);
- };
-
- //取消消费的一个回调接口 如在消费的时候队列被删除掉了
- CancelCallback cancelCallback=(consumerTag)->{
- System.out.println("消息消费被中断");
- };
- /**
- * 1、消费哪个队列
- * 2、消费成功后是否自动应答
- * 3、消费未成功回调
- * 4、消费成功回调
- */
- String s = channel.basicConsume(QUEUE_NAME, true,deliverCallback,cancelCallback);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- }
SpringAMQP是基于RabbitMQ封装的一套模板,并且还利用SpringBoot对其实现了自动装配,使用起来非常方便。
<!--AMQP依赖,包含RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1、声明交换机和队列
- @Configuration
- public class FanoutConfig {
- /**
- * 声明交换机
- * @return Fanout类型交换机
- */
- @Bean
- public FanoutExchange fanoutExchange(){
- return new FanoutExchange("itcast.fanout");
- }
-
- /**
- * 第1个队列
- */
- @Bean
- public Queue fanoutQueue1(){
- return new Queue("fanout.queue1");
- }
-
- /**
- * 绑定队列和交换机
- */
- @Bean
- public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
- return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
- }
-
- /**
- * 第2个队列
- */
- @Bean
- public Queue fanoutQueue2(){
- return new Queue("fanout.queue2");
- }
-
- /**
- * 绑定队列和交换机
- */
- @Bean
- public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
- return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
- }
- }
2、生产者
- spring:
- rabbitmq:
- host: 192.168.150.101 # 主机名
- port: 5672 # 端口
- virtual-host: / # 虚拟主机
- username: itcast # 用户名
- password: 123321 # 密码
-
- @Autowired
- private RabbitTemplate rabbitTemplate;
-
- @Test
- public void testSimpleQueue() {
- // 队列名称
- String queueName = "simple.queue";
- // 消息
- String message = "hello, spring amqp!";
- // 发送消息
- rabbitTemplate.convertAndSend(queueName, message);
- }
3、消费者
- spring:
- rabbitmq:
- host: 192.168.150.101 # 主机名
- port: 5672 # 端口
- virtual-host: / # 虚拟主机
- username: itcast # 用户名
- password: 123321 # 密码
- @RabbitListener(queues = "fanout.queue1")
- public void listenFanoutQueue1(String msg) {
- System.out.println("消费者1接收到Fanout消息:【" + msg + "】");
- }
-
- @RabbitListener(queues = "fanout.queue2")
- public void listenFanoutQueue2(String msg) {
- System.out.println("消费者2接收到Fanout消息:【" + msg + "】");
- }
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。