Rabbitmq的Federation Exchange

news2024/11/15 11:39:36

(broker 北京 ) ， (broker 深圳 ) 彼此之间相距甚远，网络延迟是一个不得不面对的问题。有一个在北京的业务(Client 北京 ) 需要连接 (broker 北京 ) ，向其中的交换器 exchangeA 发送消息，此时的网络延迟很小，(Client 北京 ) 可以迅速将消息发送至 exchangeA 中，就算在开启了 publisherconfirm 机制或者事务机制的情况下，也可以迅速收到确认信息。此时又有个在深圳的业务(Client 深圳 ) 需要向 exchangeA 发送消息，那么(Client 深圳 ) (broker 北京 ) 之间有很大的网络延迟， (Client 深圳 ) 将发送消息至 exchangeA 会经历一定的延迟，尤其是在开启了 publisherconfirm 机制或者事务机制的情况下， (Client 深圳 ) 会等待很长的延迟时间来接收(broker 北京 ) 的确认信息，进而必然造成这条发送线程的性能降低，甚至造成一定程度上的

阻塞。

解决方案

将业务 (Client 深圳 ) 部署到北京的机房可以解决这个问题，但是如果 (Client 深圳 ) 调用的另些服务都部署在深圳，那么又会引发新的时延问题，总不见得将所有业务全部部署在一个机房，那么容灾又何以实现？这里使用 Federation 插件就可以很好地解决这个问题

运行rabbitmq集群

开启插件

在每台机器上开启 federation 相关插件

rabbitmq-plugins enable rabbitmq_federation
rabbitmq-plugins enable rabbitmq_federation_management

原理图

upstream：表示上游交换机

downstream：表示下游交换机

同步数据的时候是从上游交换机传输给下游交换机，图中node1作为上游传输数据给node2下游交换机，反之如果node2要传输数据给node1，node2则为上游交换急、node1为下游交换机，传输给下游交换机时，下游交换机必须要有fed_exchange才能接收上游的数据

创建fed_exchange

为node2创建fed_exchange和队列

public static void main(String[] args) throws IOException, TimeoutException {
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.2.3");
        factory.setUsername("admin");
        factory.setPassword("123");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(FED_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.queueDeclare("node2_queue",true,false,false,null);
        channel.queueBind("node2_queue",FED_EXCHANGE,"routekey");
    }