rabbitmq 安装事件交换器插件 rabbitmq-event-exchange及使用

硅谷探秘者 Md rabbitmq,springboot 1575 0 0

客户端connection、channels、queues、consumers和系统其它部分自动生成的事件。例如,当一个connection被接受,虚拟主机通过了验证授权,将会发送一个connection_created事件,当一个connection关闭或者由于其它原因失败,将会发送一个connection_closed事件。RabbitMQ提供了一个最小的事件通知机制向RabbitMQ客户端公开。

一、rabbitmq-event-exchange 插件

rabbitmq-event-exchange是一个消费内部事件并且重新发送到一个topic exchange的插件,因此可以展示事件给客户端应用程序。

为了消费事件,应用程序需要声明一个队列,并绑定到一个系统指定的交换器去消费消息。

插件在默认的虚拟主机上声明了一个topic类型的exchange(交换器)amq.rabbitmq.event。所有的事件都会发送到这个exchange上并绑定一个路由键,比如,exchange.created、binding.deleted。所以你可以只订阅你关注的事件。

交换器的行为类似amq.rabbitmq.log,所有的信息都发布到这里,如果用户没有经过授权,你可以拒绝它们访问。

每个事件都有与之关联的各种属性,它们被转换成AMQP 0-9-1数据编码并插入到消息头中。消息的正文始终为空。

启动插件

rabbitmq-plugins enable rabbitmq_event_exchange

关闭插件

rabbitmq-plugins disable rabbitmq_event_exchange

二、事件(Events)

RabbitMQ和相关插件通过routing keys发送事件:

RabbitMQ Broker

  • Queue, Exchange and Binding events:

    • queue.deleted
    • queue.created
    • exchange.created
    • exchange.deleted
    • binding.created
    • binding.deleted
  • Connection and Channel events:

    • connection.created
    • connection.closed
    • channel.created
    • channel.closed
  • Consumer events:

    • consumer.created
    • consumer.deleted
  • Policy and Parameter events:

    • policy.set
    • policy.cleared
    • parameter.set
    • parameter.cleared
  • Virtual host events:

    • vhost.created
    • vhost.deleted
  • User related events:

    • user.authentication.success
    • user.authentication.failure
    • user.created
    • user.deleted
    • user.password.changed
    • user.password.cleared
    • user.tags.set
  • Permission events:

    • permission.created
    • permission.deleted

Shovel Plugin

  • Worker events:

    • shovel.worker.status
    • shovel.worker.removed

Federation Plugin

  • Link events:

    • federation.link.status
    • federation.link.removed

三、集群中如何使用该插件

在集群中每个节点都要开启此插件,否则amq.rabbitmq.event交换器不可以正常的被创建。

四、Springboot监听事件

添加rabbitmq配置

spring:
  rabbitmq:
    host: 192.168.1.101
    port: 5672
    username: root
    password: root

定义事件监听器 RabbitBrokerEventListener


import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.BrokerEventListener;

import java.util.Iterator;
import java.util.Map;
import java.util.Set;
/**
 * @author : Jiajiajia
 * @createDate : 2023/7/11 20:20
 * @description : 事件监听器
 */
public class RabbitBrokerEventListener extends BrokerEventListener {
    public RabbitBrokerEventListener(ConnectionFactory connectionFactory, String... eventKeys) {
        super(connectionFactory, eventKeys);
    }

    @Override
    public void onMessage(Message message) {
        System.out.println("");
        System.out.println("--------------------------------");
        System.out.println("");
        final String receivedRoutingKey = message.getMessageProperties().getReceivedRoutingKey();
        System.out.println(receivedRoutingKey);
        System.out.println();
        final Map<String, Object> headers = message.getMessageProperties().getHeaders();
        final Set<Map.Entry<String, Object>> entries = headers.entrySet();
        final Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
        while (iterator.hasNext()) {
            // 打印header
            final Map.Entry<String, Object> next = iterator.next();
            System.out.println(next.getKey() + ":" + next.getValue());
        }
        super.onMessage(message);
    }
}

配置监听器 RabbitConfig


import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * @author : Jiajiajia
 * @createDate : 2023/7/11 20:20
 * @description :  配置事件监听器
 */
@Configuration
public class RabbitConfig {

    @Bean
    public RabbitBrokerEventListener brokerEventListener(AbstractConnectionFactory connectionFactory){
        // eventKeys 是要监听的主题
        return new RabbitBrokerEventListener(connectionFactory,"user.#", "channel.#", "queue.#", "connection.#", "consumer.#");
    }
}

评论区
请写下您的评论...
暂无评论...
猜你喜欢
official 854 续介绍在rabbitmqrabbitmq_delayed_message_exchange使的方法。下载与rabbitmq版本相符的,下载地址:https://github
rabbitmq 887 rabbitmq_auth_backend_http一、rabbitmq中启rabbitmq_auth_backend_http二、修改配置文三、开启验证缓存功能一、rabbitmq中启
linux 1047 一、erlang二、socat三、rabbitmq四、启动服务五、开启户远程访问六、rabbitmq服务常命令一、erlang文下载地址:https://github.com
official 867 上一篇《(mq)rabbitmq延时队列实现延时消息1》文章中介绍了rabbitmq延时队列。本编将继续结合代码来实现延时队列(基于springboot项目)。下方所有源代码均已上传
official 764 下载rabbitmq镜像dockerpullrabbitmq:management启动rabbitmqdockerrun-d-p5672:5672-p15672:15672
official 743 RabbitMQ作为一个消息队列提供一个通的消息发送和接收平台,并且保证消息在传输过程中的全可靠。消息(Message)由Client发送,RabbitMQ接收到消息之后通过机转发到对应的队
official 784 代码已上传github,地址:https://github.com/18438301593/rabbitmq
official 811 是通过回调方法,无论消息发送到机是否成功,或是否成功路由到队列,都会通过回调方法来感知这些。配置文修改spring: rabbitmq: publisher-confirm-type:corr
归档
2018-11  12 2018-12  33 2019-01  28 2019-02  28 2019-03  32 2019-04  27 2019-05  33 2019-06  6 2019-07  12 2019-08  12 2019-09  21 2019-10  8 2019-11  15 2019-12  25 2020-01  9 2020-02  5 2020-03  16 2020-04  4 2020-06  1 2020-07  7 2020-08  13 2020-09  9 2020-10  5 2020-12  3 2021-01  1 2021-02  5 2021-03  7 2021-04  4 2021-05  4 2021-06  1 2021-07  7 2021-08  2 2021-09  8 2021-10  9 2021-11  16 2021-12  14 2022-01  7 2022-05  1 2022-08  3 2022-09  2 2022-10  2 2022-12  5 2023-01  3 2023-02  1 2023-03  4 2023-04  2 2023-06  3 2023-07  4 2023-08  1 2023-10  1 2024-02  1 2024-03  1 2024-04  1
标签
算法基础 linux 前端 c++ 数据结构 框架 数据库 计算机基础 储备知识 java基础 ASM 其他 深入理解java虚拟机 nginx git 消息中间件 搜索 maven redis docker dubbo vue 导入导出 软件使用 idea插件 协议 无聊的知识 jenkins springboot mqtt协议 keepalived minio mysql ensp 网络基础 xxl-job rabbitmq haproxy srs 音视频 webrtc javascript
目录
没有一个冬天不可逾越,没有一个春天不会来临。最慢的步伐不是跬步,而是徘徊,最快的脚步不是冲刺,而是坚持。