SpringBoot中使用RabbitMQ-阿里云开发者社区

一、搭建环境

1、引入环境依赖

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.5.RELEASE</version>
    </parent>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>

2、配置配置文件（properties）

server.port=9090
server.servlet.context-path=/rabbitMQ
#设置RabbitMQ的IP地址
spring.rabbitmq.host=192.168.5.128
#设置rabbitmq服务器连接端口
spring.rabbitmq.port=5672
#设置rabbitmq服务器虚拟主机
spring.rabbitmq.virtual-host=yingxue
#设置rabbitmq服务器用户名
spring.rabbitmq.username=whj
#设置rabbitmq服务器密码
spring.rabbitmq.password=563135

3、测试rabbitTemplate

  @Test
    public void testConnection(){
        //结果： org.springframework.amqp.rabbit.core.RabbitTemplate@2dfe5525
        System.out.println(rabbitTemplate);
    }
}

二、hello world模型

在上图的模型中，有以下概念：

P(Provider)：生产者，也就是要发送消息的程序

C(Consumer)：消费者：消息的接收者，会一直等待消息到来。

queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

1、开发生产者

/**
 * @author 王恒杰
 * @date 2022/1/23 14:10
 * @Description:
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitMqApplication.class)
public class Provider {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * Hello模型的生产者
     */
    @Test
    public void TestProviderHello(){
    //    发送消息 参数：             队列名称   发送内容
        rabbitTemplate.convertAndSend("hello_王恒杰","Hello模型发送信息");
        System.out.println("发送成功");
    }
}

2、开发消费者

@Component
//               Declare:声明              队列名               队列持久化 durable:持久耐用  自动删除
@RabbitListener(queuesToDeclare = @Queue(value = "hello_王恒杰",durable = "true",autoDelete = "true"))
public class HelloConsumer {
    /**
     * Handler:处理者
     * @param message
     */
    @RabbitHandler
    public void HelloConsumer(String message){
        System.out.println("消费者："+message);
    }
}

4、测试结果

三、work模型

work 模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会

被重复执行的。

角色：

P：生产者：任务的发布者

C1：消费者-1，领取任务并且完成任务，睡眠1000毫秒

C2：消费者-2：领取任务并完成任务

1、开发生产者

  /**
     * Work模型的生产者
     */
    @Test
    public void  TestProviderWork(){
        for (int i = 1; i <= 21; i++) {
        //    发送消息 参数：队列名称 发送内容
            rabbitTemplate.convertAndSend("work_王恒杰"," Work模型发送王恒杰这个消息"+i+"次！");
        }
        System.out.println("发送成功");
    }

2、开发消费者

  /**
     * Work模型的生产者
     */
    @Test
    public void  TestProviderWork(){
        for (int i = 1; i <= 21; i++) {
        //    发送消息 参数：队列名称 发送内容
            rabbitTemplate.convertAndSend("work_王恒杰"," Work模型发送王恒杰这个消息"+i+"次！");
        }
        System.out.println("发送成功");
    }

3、测试结果公平调度

说明:默认在Spring AMQP实现中Work这种方式就是公平调度,如果需要实现能者多劳需要额外配置

4、能者多劳实现

能者多劳配置以下配置

#能者多劳配置以下配置    每个消费者每次可以消费一个
spring.rabbitmq.listener.simple.prefetch=1

四、Fanout 广播模型

在广播模式下，消息发送流程是这样的：

可以有多个消费者

每个消费者有自己的queue（队列）

每个队列都要绑定到Exchange（交换机）

生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。

交换机把消息发送给绑定过的所有队列

队列的消费者都能拿到消息。实现一条消息被多个消费者消费

1、开发生产者

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitMqApplication.class)
public class Provider {
    private static final Logger log = LoggerFactory.getLogger(Provider.class);
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * Fanout广播模型的生产者
     */
    @Test
    public void  TestProviderFanout(){
        //发送消息 参数：交换机名字，路由key 发送内容
        rabbitTemplate.convertAndSend("fount2022","","SpringBoot fanout 广播类发送信息");
        log.warn("发送成功");
    }

2、开发消费者

@Component
public class FanoutConsumer {
    private static final Logger log = LoggerFactory.getLogger(FanoutConsumer.class);
    /**
     * 消费者1：
     * receive:接收
     * bindings配置绑定 @QueueBinding(
     * value队列@Queue,
     * exchange交换机@Exchange(name交换机名 key路由key type交换机类型)
     * durable="是否持久化"
     * autoDelete="是否自动删除")
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,//创建临时队列
            exchange = @Exchange(name = "fount2022", type = "fanout")))
    public void receive1(String message) {
        log.warn("message1=" + message);
    }
    /**
     * 消费者2
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,
            exchange = @Exchange(name = "fount2022", type = "fanout")))
    public void receive2(String message) {
        log.warn("message2=" + message);
    }
}

3、测试结果

五、Route 路由模型

图解：

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。

X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列

C1：消费者，其所在队列指定了需要routing key 为 error 的消息

C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

1、开发生产者

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitMqApplication.class)
public class Provider {
    private static final Logger log = LoggerFactory.getLogger(Provider.class);
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * Route 路由模型的生产者  direct直接的
     */
    @Test
    public void testProviderRoute() {
        rabbitTemplate.convertAndSend("route_direct_王恒杰",
                                      "error",
                                   "王恒杰通过路由直接发送发送错误的日志信息error");
        log.warn("发送成功");
    }
    }

2、开发消费者

@Component
public class RedirectConsumer {
    private static final Logger log = LoggerFactory.getLogger(RedirectConsumer.class);
    /**
     * 消费者1
     * @param message
     */
    @RabbitListener(bindings = {@QueueBinding(
            value = @Queue,
            key = { "info","warn"},
            exchange = @Exchange(type = "direct", name = "route_direct_王恒杰"))})
    public void RedirectConsumer1(String message) {
        log.warn("message1=" + message);
    }
    /**
     * 消费者2
     * @param message
     */
    @RabbitListener(bindings = {@QueueBinding(
            value = @Queue,
            key = {"error", "warn"},
            exchange = @Exchange(type = "direct", name = "route_direct_王恒杰"))})
    public void RedirectConsumer2(String message) {
        log.warn("message2=" + message);
    }
}

3、测试结果

消费者1和消费者2都含有error,则都打印信息

消费者1没有error,消费者2含有error,则消费者2打印信息

六、Topic 订阅模型(动态路由模型)

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！这种模型Routingkey一般都是由一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符

# 通配符
  * (star) can substitute for exactly one word.    匹配不多不少恰好1个词
  # (hash) can substitute for zero or more words.  匹配一个或多个词
audit.*   只能匹配 audit.irs
audit.#   可匹配audit.irs.corporate 或者 audit.irs 等

1、开发生产者

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitMqApplication.class)
public class Provider {
    private static final Logger log = LoggerFactory.getLogger(Provider.class);
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testProviderTopic(){
        rabbitTemplate.convertAndSend("topics","user.save.findAll","用户查询所有");
        log.warn("发送成功");
    }
}

2、开发消费者

@Component
public class TopicConsumer {
    private static final Logger log = LoggerFactory.getLogger(TopicConsumer.class);
    @RabbitListener(bindings = {@QueueBinding(
            value = @Queue,
            //key = {"user.*"},   如果是user.* 则消费者1不能打印信息
            key = {"user.save.*"},  //如果是user.save.* 这可以打印
            exchange = @Exchange(type = "topic",name = "topics")
    )})
    public void TopicConsumer1(String message) {
        log.warn("message1=" + message);
    }
    @RabbitListener(bindings = {@QueueBinding(
            value = @Queue,
            key = {"user.#"},  //如果是user.# 可以匹配user.save 也可以匹配user.save.findAll
            exchange = @Exchange(type = "topic",name = "topics")
    )})
    public void TopicConsumer2(String message) {
        log.warn("message2=" + message);
    }
}