安科网

RabbitMQ 第六课 springboot整合rabbitmq

cj0

cj0

2020-06-21

关注 关注

导入pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<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 https://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.3.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.everjiankang</groupId>
    <artifactId>rabbitmqDemo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>rabbitmqDemo</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-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
        </dependency>
        
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        
        <!--  导入配置文件处理器，配置文件进行绑定就会有提示 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
        </dependency>
        
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.71</version>
        </dependency>
        
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
        </dependency>
        
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

pom.xml

配置生产者

第一步：配置application.properties

spring.rabbitmq.addresses=127.0.0.1:5672
spring.rabbitmq.username=xiaochao
spring.rabbitmq.password=root
spring.rabbitmq.virtual-host=/
spring.rabbitmq.connection-timeout=15000

#生产端配置
#开启发送确认,此配置在Springboot2.3.0版本中已经@Deprecated了，默认就是
# spring.rabbitmq.publisher-confirms=true
#
spring.rabbitmq.publisher-confirm-type=simple
#开启发送失败退回
spring.rabbitmq.publisher-returns=true
#开启执行return回调
spring.rabbitmq.template.mandatory=true

第二步：开发启动类Application.java

package com.dwz.springboot;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

第三步：加入实体类order

package com.everjiankang.dependency.model;

import java.io.Serializable;

public class Order implements Serializable {
    /**
     * 
     */
    private static final long serialVersionUID = 1L;
    
    private String id;
    private String name;
    
    public Order() {
    }

    public Order(String id, String name) {
        super();
        this.id = id;
        this.name = name;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    } 
}

第四步：开发消息发送方法

package com.everjiankang.dependency.springboot.producer;

import java.util.Map;

import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import com.everjiankang.dependency.model.Order;

@Component
public class RabbitmqSender {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    /**
     * publisher-confirms,实现一个监听器用于监听broker端给我们返回的确认请求：
     * RabbitTemplate.ConfirmCallback
     * 
     * publisher-returns，保证消息对broker端是可达的，如果出现路由键不可达的情况，
     *     则使用监听器对不可达的消息进行后续的处理，保证消息的路由成功：
     * RabbitTemplate.ReturnCallback
     * 
     * 注意：在发送消息的时候对template配置mandatory=true保证ReturnCallback监听有效
     * 生产端还可以配置其他属性，比如发送重试，超时时间、次数、间隔等。
     */
    final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.err.println("confirm correlationData:" + correlationData);
            System.err.println("confirm ack:" + ack);
            if(!ack) {
                System.err.println("异常处理。。。");
            }
        }
    };
    
    final ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
        
        @Override
        public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
                String exchange, String routingKey) {
            System.err.println("return exchange:" + exchange);
            System.err.println("return routingKey:" + routingKey);
            System.err.println("return replyCode:" + replyCode);
            System.err.println("return replyText:" + replyText);
        }
    };
    
    public void send(Object message, Map<String, Object> properties) throws Exception {
        MessageHeaders mhs = new MessageHeaders(properties);
        Message msg = MessageBuilder.createMessage(message, mhs);
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnCallback(returnCallback);
        CorrelationData correlationData = new CorrelationData();
        //实际消息的唯一id
        correlationData.setId("dwz123456");//id + 时间戳 （必须是全局唯一的）
        rabbitTemplate.convertAndSend("exchange-1", "spring.abc", msg, correlationData);
    }
    
    public void sendOrder(Order order) throws Exception {
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnCallback(returnCallback);
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId("zheaven123456");//id + 时间戳 （必须是全局唯一的）
        rabbitTemplate.convertAndSend("exchange-2", "springboot.def", order, correlationData);
    }
}

第五步：测试发送消息

package com.dwz.springboot;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import com.dwz.springboot.entity.Order;
import com.dwz.springboot.producer.RabbitSender;
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringbootTest {
    @Autowired
    private RabbitSender send;
    
    private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS");
    
    @Test
    public void testSender1() throws Exception {
        Map<String, Object> properties = new HashMap<String, Object>();
        properties.put("number", "12345");
        properties.put("send_time", simpleDateFormat.format(new Date()));
        send.send("Hello Rabbirmq For springboot!", properties);
        
    }
    
    @Test
    public void testSender2() throws Exception {
        Order order = new Order("001", "第一个订单");
        send.sendOrder(order);
    }
}

配置消费者

第一步：配置application.properties

spring.rabbitmq.addresses=127.0.0.1:5672
spring.rabbitmq.username=xiaochao
spring.rabbitmq.password=root
spring.rabbitmq.virtual-host=/vhost_dwz
spring.rabbitmq.connection-timeout=15000

#消费端配置
#配置手工确认模式，用于ACK的手工处理，这样我们可以保证消息的可靠性送达，或者在消费端消费失败的时候可以做到重回队列、根据业务记录日志等处理
spring.rabbitmq.listener.simple.acknowledge-mode=manual
#可以设置消费端的监听个数和最大个数，用于控制消费端的并发情况
spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10

#消费端binding配置 ,以订单为例
spring.rabbitmq.listener.order.queue.name=queue-2
spring.rabbitmq.listener.order.queue.durable=true
spring.rabbitmq.listener.order.exchange.name=exchange-2
spring.rabbitmq.listener.order.exchange.durable=true
spring.rabbitmq.listener.order.exchange.type=topic
spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
spring.rabbitmq.listener.order.key=springboot.*

第二步：开发启动类Application.java

package com.dwz.springboot;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Application {
    
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

第三步：添加主配置类

package com.dwz.springboot;

import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

@Configuration
@ComponentScan({"com.dwz.springboot.*"})
public class MainConfig {

}

第四步：添加消息处理方法

package com.dwz.springboot.consumer;

import java.util.Map;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import com.dwz.springboot.entity.Order;
import com.rabbitmq.client.Channel;

@Component
public class RabbitReceiver {
    
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(
                        name = "queue-1",
                        durable = "true"
                    ),
            exchange = @Exchange(
                        name = "exchange-1",
                        durable = "true",
                        type = "topic",
                        ignoreDeclarationExceptions = "true"
                    ),
            key = "springboot.*"
            )
    )
    @RabbitHandler
    public void onMessage(Message message, Channel  channel) throws Exception {
        System.err.println("------------------------------------");
        System.err.println("消费端Payload:" + message.getPayload());
        Long deliverytag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        //手工ACK
        channel.basicAck(deliverytag, false);
    }
    
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(
                        name = "${spring.rabbitmq.listener.order.queue.name}",
                        durable = "${spring.rabbitmq.listener.order.queue.durable}"
                    ),
            exchange = @Exchange(
                        name = "${spring.rabbitmq.listener.order.exchange.name}",
                        durable = "${spring.rabbitmq.listener.order.exchange.durable}",
                        type = "${spring.rabbitmq.listener.order.exchange.type}",
                        ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"
                    ),
            key = "${spring.rabbitmq.listener.order.key}"
            )
    )

    @RabbitHandler
    //将Message用@Payload和@Headers拆分
    //@Payload表示body里面的信息，@Headers表示properties的信息
    public void onOrderMessage(@Payload Order order, Channel  channel,
            @Headers Map<String, Object> headers) throws Exception {
        System.err.println("------------------------------------");
        System.err.println("消费端Order:" + order.getId());
        Long deliverytag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        //手工ACK
        channel.basicAck(deliverytag, false);
    }
}

 参考blog：

https://www.cnblogs.com/zheaven/p/11915480.html

 参考视频：

https://www.bilibili.com/video/BV1Fz411B77A?p=34

rabbitmq

cj0

cj0

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

RabbitMQ如何保证消息的可靠投递？

String message = "this is info message " + i;autoAck=false: RabbitMQ会等待消费者显示回复确认消息后才从内存中移出消息。deliveryTag: 用来标识信道中投递的消息

zhuxue 2020-10-14

RabbitMQ七战Kafka，差异立现！

作为一个有丰富经验的微服务系统架构师，经常有人问我，“应该选择RabbitMQ还是Kafka？”。基于某些原因， 许多开发者会把这两种技术当做等价的来看待。的确，在一些案例场景下选择RabbitMQ还是Kafka没什么差别，但是这两种技术在底层实现方面是有

shenzhenzsw 2020-10-09

RabbitMQ是如何确定消息是否投递到队列中的

在使用RabbitMQ消息中间件时，因为消息的投递是异步的，默认情况下，RabbitMQ会删除那些无法路由的消息。为了能够检出消息是否顺利投递到队列，我们需要相应的处理机制。今天就来验证一下相关的验证机制。RabbitMQ消息会先到达指定的交换机，然后由交

shyoldboy 2020-09-27

c# rabbitmq 简单收发消息的示例代码

//设置prefetchCount : 1来告知RabbitMQ，在未收到消费端的消息确认时，不再分发消息，也就确保了当消费端处于忙碌状态时，不再分配任务。channel.BasicAck;// 消息确认信号，确认后该消息将从队列里移除

leihui00 2020-09-16

Docker 安装Redis、MySql、ELK、RabbitMq

docker run -d --restart always --name kibana --net somenetwork -p 5601:5601 kibana:7.1.1

lishijian 2020-08-17

《RabbitMQ》如何保证消息不被重复消费

消息重复的原因有两个：1.生产时消息重复，2.消费时消息重复。由于生产者发送消息给MQ，在MQ确认的时候出现了网络波动，生产者没有收到确认，实际上MQ已经接收到了消息。生产者中如果消息未被确认，或确认失败，我们可以使用定时任务+来进行消息重试。@Sched

程序员伊成 2020-08-06

RabbitMQ都写了，RocketMQ怎么能落下？

整体架构最近看到了我在Github上写的rabbitmq-examples陆续被人star了，就想着写个rocketmq-examples。对rabbitmq感兴趣的小伙伴可以看我之前的文章。下面把RocketMQ的各个特性简单介绍一下，这样在用的时候心里

ljcsdn 2020-07-27

RabbitMQ —— 介绍

AMQP，即 AMQP，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开发标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端中间件不同产品，不同的开发语言等条件的限制，Erlang 中的实现有 Rab

waitzkj 2020-07-25

mq如何处理消息丢失

　　生产者将数据发送到rabbitmq的时候，可能因为网络问题导致数据就在半路给搞丢了。但是问题是，开始rabbitmq事务机制，基本上吞吐量会下来，因为太耗性能。如果rabbitmq没能处理这个消息，会回调你一个nack接口，告诉你这个消息接收失败，你可

powrexly 2020-07-20

ubuntu下快速安装rabbitmq

安装rabbitmqsudo apt-get updatesudo apt-get upgradesudo apt-get install rabbitmq-server. 当我们在创建用户时，会指定用户能访问一个虚拟机，并且该用户只能访问该虚拟机下的队列

liym 2020-07-20

深入浅出 RabbitMQ

基于 ErLang 语言开发有高可用高并发的优点，适合集群。有消息确认机制和持久化机制，可靠性高。消息队列提供了 FIFO 的处理机制，具有缓存消息的能力。在 RabbitMQ 中，队列消息可以设置为持久化，临时或者自动删除。如果是临时的队列，Queue

zhoucheng0 2020-07-19

Docker-compose常用软件部署(mysql、redis、rabbitmq)

废话不多说，下面就直接给出部署文件已经启动脚本，这个会持续更新。以下的部署方式均是单例部署。高可用或者其他部署方案我会单独写相应的文章。不指定则会启动失败。filterName=`docker network ls | grep $network_name

shenzhenzsw 2020-07-18

Rabbitmq的使用四_Java Client方式使用Topic模式

topic交换机发送给队列的消息，通过routingkey进行匹配。　　* 号只能代替一个单词。　　# 号可以替换零个或多个单词。在这个例子中，我们将发送所有描述动物的信息。消息将与一个routingkey一起发送，routingkey由三个单词(两个点)

woaishanguosha 2020-07-18

rabbitmq 延时队列

‘queueName‘ => ‘test_cache_queue‘."_".$time,‘routeKey‘ => ‘test_cache_route‘."_".$time,‘host‘ => ‘1

waitzkj 2020-07-18

SpringBoot与RabbitMQ整合

消息代理和目的地。当消息发送者发送消息以后，将由消息代理接管，消息代理保证消息传递到指定目的地。–消息只有唯一的发送者和接受者，但并不是说只能有一个接收者。AMQP定义了wire-level层的协议标准；天然具有跨平台、跨语言特性。绑定，用于消息队列和交换

zhoucheng0 2020-07-08

RabbitMQ使用

那就表示一个工作队列，而工作队列的最大特征在于：若干个消费者一起完成工作。

waitzkj 2020-07-05

RabbitMQ 3.8.5版本 windows安装和web管理界面登陆

RabbitMQ默认安装路径为：C:\Program Files\RabbitMQ Server\rabbitmq_server-3.8.5

json0000 2020-07-04

rabbitmq单机入门安装教程(ubuntu环境 亲测有效)

第一步:使用xz -d rabbitmq-server-generic-unix-3.7.26.tar.xz 转换为tar格式的文件。第二步:使用 tar -xvf rabbitmq-server-generic-unix-3.7.26.tar 解压文件。

NVEFLY 2020-07-04

springcloud-消息总线Bus(九)

SpringCloud Bus配合Config使用可以实现配置的动态刷新.Bus是用来将分布式系统的节点与轻量级消息系统链接起来的框架, 它整合了Java的事件处理机制和消息中间件的功能.Bus能管理和传播分布式系统间的消息, 就像一个分布式执行器, 可用

OnMyHeart 2020-07-04

Rabbitmq安装

RabbitMQ在安装好后，可以访问http://ip地址:15672 ；其自带了guest/guest的用户名和密码；如果需要创建自定义用户；那么也可以登录管理界面后，如下操作：。可登陆管理控制台, 同时可以对policy进行管理。仅可登陆管理控制台，无

OnMyHeart 2020-06-28
cj0

cj0

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号