新鲜出炉这是全网讲的最详细的springboot整合消息服务了吧建议收藏

4次阅读

共计 8872 个字符，预计需要花费 23 分钟才能阅读完成。

ActiveMq 是 Apache 提供的开源音讯零碎采纳 java 实现，

很好地反对 JMS（Java Message Service, 即 Java 音讯服务）标准

ActiveMq 装置：http://activemq.apache.org/co… 在官网下载安装对应的版本

下载实现后解压就能够应用

ActiveMq 默认的端口号是 8161，用户名和明码都是 admin 在本机能够应用 http://localhost:8161 去拜访

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

spring.activemq.broker-url=tcp://localhost:61616
#如果是点对点（queue），那么此处默认应该是 false，如果公布订阅，那么肯定设置为 true
spring.activemq.packages.trust-all=true
spring.activemq.user=admin
spring.activemq.password=admin

@Component
public class QueueBean{
    // 创立一个队列实例
    @Bean
    Queue queue(){
        // 这里设置的音讯是队列的名称
        return new ActiveMQQueue("hello.javaboy");
    }
}

@Component
public class JmsComponent{
    //springboot 提供的音讯模板
    @Autowired
    JmsMessagingTemplate jmsMessagingTemplate;
    // 本人创立的队列实例
    @Autowired
    Queue queue;
    /**
     * 发送音讯
     * @param message
     */
    public void send(Message message){jmsMessagingTemplate.convertAndSend(this.queue,message);
    }
    /**
     * 接管音讯
     * @param message
     */
    // 示意监听该队列名称发来的音讯
    @JmsListener(destination = "hello.javaboy")
    public void readMessage(Message message){System.out.println(message);
    }

}

public class Message implements Serializable {
    private String content;// 音讯主体
    private Date sendDate;// 音讯发送的工夫
    // 省略 get、set、tostring 办法
}

在测试类中注入 JmsComponent 调用 send() 办法进行音讯的转发

@SpringBootTest
class ActivemqApplicationTests {
    @Autowired
    JmsComponent jmsComponent;
    @Test
    void contextLoads() {Message message = new Message();
        message.setContent("hello activeMq");
        message.setSendDate(new Date());
        jmsComponent.send(message);
    }
}

首先启动我的项目，在运行测试类进行音讯发送：

控制台会打印消息内容：

rabbitmq 装置比拟繁琐，这里应用 docker 容器进行装置，docker 装置十分不便，一条命令全副搞定

通过 docker 装置 rabbitmq

-P(大 p) 示意主动映射到主机端口

docker run -d --hostname my-rabbitmq --name some-rabbitmq -P rabbitmq:3-management

首先导入依赖

 <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>

编写配置文件：

# 配置 rabbitMQ
spring.rabbitmq.host=localhost
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.port=32771

直连交换机：Direct exchange

扇形交换机：Fanout exchange

主体交换机：Topic exchange

首部交换机：Headers exchange

上面别离介绍 4 中替换模式：

//Direct 策略（只转发给 routingKey 相匹配的用户）@Configuration
public class RabbitDirectConfig {
    public final static String DIRECTNAME = "javaboy-direct";
    // 音讯队列
    @Bean
    Queue queue(){
        //name 值为队列名称，routingKey 会与他进行匹配
        return new Queue("hello.RabbitMQ");
    }
    @Bean
    Queue queue1(){return new Queue("hello.RabbitMQ1");
    }
    @Bean
    DirectExchange directExchange(){
        // 第一个参数为 DIRECTNAME、第二个参数示意重启后是否无效，第三参数示意长时间未应用是否删除
        return new DirectExchange(DIRECTNAME,true,false);
    }
    @Bean
    Binding binding(){
        // 将队列 queue 和 DirectExchange 绑定在一起
        return BindingBuilder.bind(queue()).to(directExchange()).with("direct");
    }
    @Bean
    Binding binding1(){
        // 将队列 queue 和 DirectExchange 绑定在一起
        return BindingBuilder.bind(queue1()).to(directExchange()).with("direct");
    }

}

// 配置消费者
@Component
public class DirectReceiver {// 只监听 queue() 队列的音讯
    @RabbitListener(queues = "hello.RabbitMQ")
    public void hanlder(String msg){System.out.println("hanlder>>>"+msg);

    }
    // 只监听 queue1() 队列的音讯
    @RabbitListener(queues = "hello.RabbitMQ1")
    public void hanlder1(String msg){System.out.println("hanlder1>>>"+msg);

    }
}

测试代码：

在 springboot 的测试类中注入 RabbitTemplate(springboot 提供的 RabbitMQ 模板)

 @Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    void contextLoads() {
        // 两个参数第一个是 routingKey、第二个为音讯内容
        rabbitTemplate.convertAndSend("hello.RabbitMQ","hello RabbitMQ test");
        rabbitTemplate.convertAndSend("hello.RabbitMQ1","hello RabbitMQ test222");
    }

启动我的项目后，运行测试类能够看到只有与 routingkey 相匹配的消费者受到了对应的音讯：

2、Fanout exchange

//Fanout 策略（只有是与他绑定的队列，都会收到音讯与 routingKey 无关）@Configuration
public class RabbitFanoutConfig {
    public final static String FANOUTNAME = "javaboy-fanout";
    // 配置了两个音讯队列 queueOne 和 queueTwo
    @Bean
    Queue queueOne(){return new Queue("queue-one");
    }
    @Bean
    Queue queueTwo(){return new Queue("queue-two");
    }
    @Bean
    FanoutExchange fanoutExchange(){return new FanoutExchange(FANOUTNAME,true,false);
    }
    // 将两个队列与 FanoutExchange 绑定
    @Bean
    Binding bindingOne(){return BindingBuilder.bind(queueOne()).to(fanoutExchange());
    }
    @Bean
    Binding bindingTwo(){return BindingBuilder.bind(queueTwo()).to(fanoutExchange());
    }
}

// 配置消费者
@Component
public class FanoutReceiver {
    // 两个消费者别离监听两个不同的队列
    @RabbitListener(queues = "queue-one")
    public void hanlder1(String msg){System.out.println("FanoutReceiver:hanlder1>>>"+msg);

    }
    @RabbitListener(queues = "queue-two")
    public void hanlder2(String msg){System.out.println("FanoutReceiver:hanlder2>>>"+msg);

    }
}

@Test
    void rabbitFanout(){
        // 三个参数示意 RabbitFanoutConfig 的名称、routingkey、音讯内容
        rabbitTemplate.convertAndSend(RabbitFanoutConfig.FANOUTNAME,null,"hello fanout test");
    }

该形式与 routingkey 无关所有写 null 即可

查看输入：能够看到两个消费者都收到了音讯

topic 策略能够依据 routingKey 的规定（通配符形式）进行去匹配队列进行转发规定为.#. * 为单词，# 示意含糊匹配

例如 routingkey 为：xiaomi.# 那么带有 xiaomi. 结尾的队列都会收到该音讯

routingkey 为：#.phone.# 示意音讯的 routingKey 中带有 phone 时就会去匹配带有 phone 的队列

/topic 策略能够依据 routingKey 的规定（通配符形式）进行去匹配队列进行转发规定为 *.#.*
    //* 为单词，# 示意含糊匹配
@Configuration
public class RabbitTopicConfig {
    public final static String TOPICNAME = "javaboy-topic";

    @Bean
    TopicExchange topicExchange(){return new TopicExchange(TOPICNAME,true,false);
    }
    @Bean
    Queue xiaomi(){return new Queue("xiaomi");
    }
    @Bean
    Queue huawei(){return new Queue("huawei");
    }
    @Bean
    Queue phone(){return new Queue("phone");
    }

    @Bean
    Binding xiaomiBinding(){
        //xiaomi.#：示意音讯的 routingKey 是以 xiaomi 结尾的就会路由到 xiaomi 的队列
        return BindingBuilder.bind(xiaomi()).to(topicExchange()).with("xiaomi.#");
    }
    @Bean
    Binding huaweiBinding(){return BindingBuilder.bind(huawei()).to(topicExchange()).with("huawei.#");
    }
    @Bean
    Binding phoneBinding(){
        //#.phone.#：示意音讯的 routingKey 中带 phone 的都会路由到 phone 的队列
        return BindingBuilder.bind(phone()).to(topicExchange()).with("#.phone.#");
    }
}

@Component
public class TopicReceiver {
    // 别离监听名称为 xiaomi、huawei、phone 的队列
    @RabbitListener(queues = "xiaomi")
    public void handlerXM(String msg){System.out.println("TopicReceiver:handlerXM>>>"+msg);
    }
    @RabbitListener(queues = "huawei")
    public void handlerHW(String msg){System.out.println("TopicReceiver:handlerHW>>>"+msg);
    }
    @RabbitListener(queues = "phone")
    public void handlerPHONE(String msg){System.out.println("TopicReceiver:handlerPHONE>>>"+msg);
    }
}

@Test
    void rabbitTopic(){
        // 依据匹配规定该音讯只能被 xiaomi 的队列收到
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,"xiaomi.news","小米新闻");
        // 依据匹配规定该音讯只能被 phone 的队列收到
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,"vivo.phone","vivo 手机");
        // 依据匹配规定该音讯能够别 huawei 和 phone 两个队列收到
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,"huawei.phone","华为手机");

    }

查看输入：

能够看到 routingkey 为 huawei.phone 的音讯匹配了两个队列，其余两个都只匹配了一个队列

该模式是依据路由规定的 header 进行匹配的，在进行匹配的时候须要传入一个 map 汇合，routingkey 去匹配 map 即可中的 key value，匹配规定能够使 any 或者 all，any 示意只有蕴含任意信息就能够，all 示意所有信息都必须匹配

@Configuration
public class RabbitHeaderConfig {
    public final static String HEADERNAME = "javaboy-header";

    @Bean
    HeadersExchange headersExchange(){return new HeadersExchange(HEADERNAME,true,false);
    }
    // 别离创立两个不同 header 的队列
    @Bean
    Queue queueName(){return new Queue("name-queue");
    }
    @Bean
    Queue queueAge(){return new Queue("age-queue");
    }
    @Bean
    Binding bindingName(){Map<String,Object> map = new HashMap<>();
        map.put("name","hello");
        // 示意如果 routingKey 匹配的 map 汇合中的 key value 就会将音讯转发到对应的路由上
        return BindingBuilder.bind(queueName()).to(headersExchange()).whereAny(map).match();}

    @Bean
    Binding bindingAge(){return BindingBuilder.bind(queueAge()).to(headersExchange()).where("age").exists();}
}

@Component
public class HeaderReceiver {@RabbitListener(queues = "name-queue")
    public void handlerName(byte[] msg){System.out.println("HeaderReceiver:handlerName>>>>"+new String(msg,0,msg.length));
    }
    @RabbitListener(queues = "age-queue")
    public void handlerAge(byte[] msg){System.out.println("HeaderReceiver:handlerAge>>>>"+new String(msg,0,msg.length));
    }
}

@Test
    public void rabbitHeader(){// 设置音讯，并且设置 header，setHeader("name","hello") 别离示意 map 汇合中的 key、value
        Message nameMessage = 
            MessageBuilder.withBody("hello name".getBytes()).setHeader("name","hello").build();
        Message ageMessage =
            MessageBuilder.withBody("hello 99 age".getBytes()).setHeader("age","99").build();
        rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME,null,nameMessage);
        rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME,null,ageMessage);
    }

查看输入：

扭转 setheader 中的值查看后果：

 Message nameMessage = 
            MessageBuilder.withBody("hello name".getBytes()).setHeader("name","javaboy").build();

能够看到因为 key、value 匹配不上只打印了一条音讯。

大家看完有什么不懂的能够在下方留言探讨，也能够关注我私信问我，我看到后都会答复的。也欢送大家关注我的公众号：前程有光，金三银四跳槽面试季，整顿了 1000 多道将近 500 多页 pdf 文档的 Java 面试题材料，文章都会在外面更新，整顿的材料也会放在外面。谢谢你的观看，感觉文章对你有帮忙的话记得关注我点个赞反对一下！

正文完

spring

发表至： spring

2020-07-16

0

关于spring:详解线程池的作用及Java中如何使用线程池

关于spring:Spring中毒太深离开了Spring我居然连最基本的接口都不会写了¯\ツ¯

关于spring:Spring和SpringMVC

性能测试之Jmeter中场景设置与启动方式

Kuiper-中使用-Golang-模版-template-定制分析结果

新鲜出炉这是全网讲的最详细的springboot整合消息服务了吧建议收藏

springboot 整合 activeMq

springboot 整合 ActiveMq

1、导入依赖

2、在 properties 文件中配置 activeMq

3、编写 queue（队列）

4、创立音讯的发送者以及消费者

5、上述 Message 实体类

6、进行音讯的发送以及生产

springboot 整合 RabbitMQ

RabbitMQ 四种替换模式：

1、Direct exchange

2、配置消费者 DirectReceiver：

Fanout 策略（只有是与他绑定的队列，都会收到音讯与 routingKey 无关）

1、配置 RabbitFanoutConfig：

2、配置消费者 FanoutReceiver：

3、测试类：

3、Topic exchange

1、配置 RabbitTopicConfig：

2、配置消费者 TopicReceiver：

3、测试类：

4、Headers exchange

1、配置 RabbitHeaderConfig：

2、创立消费者 HeaderReceiver:

3、测试代码：

最初

站内搜索