[Java知识库]07_整合spring boot

整合spring boot-生产者-Fanout模式

1.新建springboot项目 选择 springweb和rabbitmq依赖

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <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>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-resources-plugin</artifactId>
                <version>3.1.0</version>
            </plugin>
        </plugins>
    </build>

2.service层

package com.tian.demo.service;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.UUID;

@Service
public class OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void makeOrder(String userid,String productid,int num){
        //1.根据商品id查询库存是否充足
        //2.保存订单
        String orderId= UUID.randomUUID().toString();
        System.out.println("订单生产成功："+orderId);

        //3.通过MQ来完成消息的分发
        //参数1：交换机 参数2：路由key/queue队列名称 参数3：消息内容
        String exchangeName="fanout_order_exchange";
        String routingKey="";
        rabbitTemplate.convertAndSend(exchangeName,routingKey,orderId);

    }
}

3.config层

package com.tian.demo.config;

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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMqConfiguration {
    //1.声明注册fanout模式的交换机
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanout_order_exchange",true,false);
    }
    //2.声明队列
    @Bean
    public Queue smsQueue(){
        return new Queue("sms.fanout.queue",true);
    }
    @Bean
    public Queue duanxinQueue(){
        return new Queue("duanxin.fanout.queue",true);
    }
    @Bean
    public Queue emailQueue(){
        return new Queue("email.fanout.queue",true);
    }
    //3.完成绑定关系 （队列和交换机完成绑定关系）
    @Bean
    public Binding smsBingding(){
        return BindingBuilder.bind(smsQueue()).to(fanoutExchange());
    }
    @Bean
    public Binding emailBingding(){
        return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
    }
    @Bean
    public Binding duanxinBingding(){
        return BindingBuilder.bind(duanxinQueue()).to(fanoutExchange());
    }
}

4.application.yml

server:
  port: 8080

#配置rabbitmq
spring:
  rabbitmq:
    username: admin
    password: admin
    virtual-host: /
    host: 192.168.1.150
    port: 5672

5.测试运行

@SpringBootTest
class DemoApplicationTests {
    @Autowired
    private OrderService orderService;
    @Test
    void contextLoads() {
        orderService.makeOrder("1","1",12);
    }
}

6.查看网页

整合spring boot-消费者-Fanout模式

1.新建springboot 勾选上springweb和rabbitmq依赖 同上

2.application.yml同上

3.service层

• FanoutDuanxinConsumer

@Service
@RabbitListener(queues = {"duanxin.fanout.queue"})
public class FanoutDuanxinConsumer {
    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("duanxin fanout--->接收到了订单信息是->"+message);
    }
}

• FanoutEmailConsumer

@Service
@RabbitListener(queues = {"email.fanout.queue"})
public class FanoutEmailConsumer {
    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("email fanout--->接收到了订单信息是->"+message);
    }
}

• FanoutSMSConsumer

@Service
@RabbitListener(queues = {"sms.fanout.queue"})
public class FanoutSMSConsumer {
    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("sms fanout--->接收到了订单信息是->"+message);
    }
}

4.主启动类运行 控制台收到消息

sms fanout--->接收到了订单信息是->2531c6ab-18be-4d5f-b880-835533c41dd4
email fanout--->接收到了订单信息是->2531c6ab-18be-4d5f-b880-835533c41dd4
duanxin fanout--->接收到了订单信息是->2531c6ab-18be-4d5f-b880-835533c41dd4

整合spring boot-Direct模式

• 配置文件写在消费端更好 启动不会报错

consumer config层

package com.tian.demo.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DirectRabbitMqConfiguration {
    //1.声明注册fanout模式的交换机
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange("direct_order_exchange",true,false);
    }
    //2.声明队列
    @Bean
    public Queue directsmsQueue(){
        return new Queue("sms.direct.queue",true);
    }
    @Bean
    public Queue directduanxinQueue(){
        return new Queue("duanxin.direct.queue",true);
    }
    @Bean
    public Queue directemailQueue(){
        return new Queue("email.direct.queue",true);
    }
    //3.完成绑定关系 （队列和交换机完成绑定关系）
    @Bean
    public Binding directsmsBingding(){
        return BindingBuilder.bind(directsmsQueue()).to(directExchange()).with("sms");
    }
    @Bean
    public Binding directemailBingding(){
        return BindingBuilder.bind(directemailQueue()).to(directExchange()).with("email");
    }
    @Bean
    public Binding directduanxinBingding(){
        return BindingBuilder.bind(directduanxinQueue()).to(directExchange()).with("duanxin");
    }
}

service层

@Service
@RabbitListener(queues = {"duanxin.direct.queue"})
public class DirectDuanxinConsumer {
    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("duanxin direct--->接收到了订单信息是->"+message);
    }
}

@Service
@RabbitListener(queues = {"email.direct.queue"})
public class DirectEmailConsumer {
    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("email direct--->接收到了订单信息是->"+message);
    }
}


@Service
@RabbitListener(queues = {"sms.direct.queue"})
public class DirectSMSConsumer {
    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("sms direct--->接收到了订单信息是->"+message);
    }
}

生产者 service层

 public void makeOrderDirect(String userid,String productid,int num){
        //1.根据商品id查询库存是否充足
        //2.保存订单
        String orderId= UUID.randomUUID().toString();
        System.out.println("订单生产成功："+orderId);

        //3.通过MQ来完成消息的分发
        //参数1：交换机 参数2：路由key/queue队列名称 参数3：消息内容
        String exchangeName="direct_order_exchange";
        rabbitTemplate.convertAndSend(exchangeName,"email",orderId);
        rabbitTemplate.convertAndSend(exchangeName,"duanxin",orderId);

    }

测试访问

@Autowired
private OrderService orderService;
@Test
void contextLoads1() {
    orderService.makeOrderDirect("1","1",12);
}

只有 邮件和短信收到服务

整合spring boot-Topic模式

• 注解模式
• consumer端service层

@Service
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "duanxin.topic.queue",durable = "true",autoDelete = "false"),
        exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
        key = "#.duanxin.#"
))
public class TopicDuanxinConsumer {
    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("duanxin topic--->接收到了订单信息是->"+message);
    }
}

@Service
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "email.topic.queue",durable = "true",autoDelete = "false"),
        exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
        key = "*.email.#"
))
public class TopicEmailConsumer {
    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("email topic--->接收到了订单信息是->"+message);
    }
}

@Service
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "sms.topic.queue",durable = "true",autoDelete = "false"),
        exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
        key = "com.#"
))
public class TopicSMSConsumer {
    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("sms topic--->接收到了订单信息是->"+message);
    }
}

producer端 service层

   public void makeOrderTopic(String userid,String productid,int num){
        //1.根据商品id查询库存是否充足
        //2.保存订单
        String orderId= UUID.randomUUID().toString();
        System.out.println("订单生产成功："+orderId);

        //3.通过MQ来完成消息的分发
        //参数1：交换机 参数2：路由key/queue队列名称 参数3：消息内容
        String exchangeName="topic_order_exchange";
        String routingKey="com.duanxin";
        // #.duanxin.#   duanxin
        // *.email.#   email
        // com.#     sms
        rabbitTemplate.convertAndSend(exchangeName,routingKey,orderId);

测试

@Autowired
private OrderService orderService;
@Test
void contextLoads2() {
    orderService.makeOrderTopic("1","1",12);
}

只有sms 和 短信接收到服务