RibbitMQ学习笔记之交换机实战（二）-阿里云开发者社区

5.5.4. 实战

public class ReceiveLogsDirect01 {
private static final String EXCHANGE_NAME = "direct_logs"; public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); String queueName = "disk";
channel.queueDeclare(queueName, false, false, false, null); channel.queueBind(queueName, EXCHANGE_NAME, "error");
System.out.println("等待接收消息 ........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8");
message="接收绑定键:"+delivery.getEnvelope().getRoutingKey()+",消息:"+message; File file = new File("C:\\work\\rabbitmq_info.txt"); FileUtils.writeStringToFile(file,message,"UTF-8");
System.out.println("错误日志已经接收");
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
}
}

public class ReceiveLogsDirect02 {
private static final String EXCHANGE_NAME = "direct_logs"; public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel();
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); String queueName = "console";
channel.queueDeclare(queueName, false, false, false, null); channel.queueBind(queueName, EXCHANGE_NAME, "info"); channel.queueBind(queueName, EXCHANGE_NAME, "warning");
System.out.println("等待接收消息 ........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" 接收绑定键 :"+delivery.getEnvelope().getRoutingKey()+", 消息:"+message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
}
}

public class EmitLogDirect {
private static final String EXCHANGE_NAME = "direct_logs"; public static void main(String[] argv) throws Exception {
try (Channel channel = RabbitUtils.getChannel())
{ channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
//创建多个 bindingKey
Map<String, String> bindingKeyMap = new HashMap<>(); bindingKeyMap.put("info","普通 info 信息"); bindingKeyMap.put("warning","警告 warning 信息"); bindingKeyMap.put("error","错误 error 信息");
//debug 没有消费这接收这个消息 所有就丢失了
bindingKeyMap.put("debug","调试 debug 信息");
for (Map.Entry<String, String> bindingKeyEntry: bindingKeyMap.entrySet()){ String bindingKey = bindingKeyEntry.getKey();
String message = bindingKeyEntry.getValue(); channel.basicPublish(EXCHANGE_NAME,bindingKey, null,
message.getBytes("UTF-8"));
System.out.println("生产者发出消息:" + message);
}
}
}

自己写的

package com.atguigu.rabbitmq.six;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ReceiveLogsDirect01 {
    public static final String EXCHANGE_NAME="direct_logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqutils.getChannel();
        //声明一个交换机
        /**
         * 1.交换机的名字
         * 2.类型 fanout 主要练习的就是他
         */
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        //声明一个队列
        channel.queueDeclare("console",false,false,false,null);
        channel.queueBind("console",EXCHANGE_NAME,"info");
        channel.queueBind("console",EXCHANGE_NAME,"warning");
        //接受消息
        DeliverCallback deliverCallback = (var1, var2) ->
        {
            System.out.println("ReceiveLogsDirect01控制台打印接口道的消息：-[]" + new String(var2.getBody(), "utf-8"));
        };
        //消费者取消消息时回调的接口
        channel.basicConsume("console", true, deliverCallback, var1 -> {
        }, null);
    }
}

package com.atguigu.rabbitmq.six;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ReceiveLogsDirect02 {
    public static final String EXCHANGE_NAME="direct_logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqutils.getChannel();
        //声明一个交换机
        /**
         * 1.交换机的名字
         * 2.类型 fanout 主要练习的就是他
         */
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        //声明一个队列
        channel.queueDeclare("disk",false,false,false,null);
        channel.queueBind("disk",EXCHANGE_NAME,"error");
        //接受消息
        DeliverCallback deliverCallback = (var1, var2) ->
        {
            System.out.println("ReceiveLogsDirect02控制台打印接口道的消息：-[]" + new String(var2.getBody(), "utf-8"));
        };
        //消费者取消消息时回调的接口
        channel.basicConsume("disk", true, deliverCallback, var1 -> {
        }, null);
    }
}

package com.atguigu.rabbitmq.six;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;
public class DirectLogs {
    //交换机的名字
    public static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqutils.getChannel();
        //交互机
        // channel.exchangeDeclare(EXCHANGE_NAME,"fauout");
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            String message = scanner.next();
           // channel.basicPublish(EXCHANGE_NAME, "error", null, message.getBytes("utf-8"));
          //  channel.basicPublish(EXCHANGE_NAME, "info", null, message.getBytes("utf-8"));
            channel.basicPublish(EXCHANGE_NAME, "warning", null, message.getBytes("utf-8"));
            System.out.println("生产者发出的消息是什么" + message);
        }
    }
}

5.6. Topics 主题交换机

5.6.1. 之前类型的问题

在上一个小节中，我们改进了日志记录系统。我们没有使用只能进行随意广播的 fanout 交换机，而是使用了 direct 交换机，从而有能实现有选择性地接收日志。

尽管使用direct 交换机改进了我们的系统，但是它仍然存在局限性-比方说我们想接收的日志类型有 info.base 和 info.advantage，某个队列只想 info.base 的消息，那这个时候direct 就办不到了。这个时候就只能使用 topic 类型

5.6.2. Topic 的要求

发送到类型是 topic 交换机的消息的 routing_key 不能随意写，必须满足一定的要求，它必须是一个单

词列表，以点号分隔开。这些单词可以是任意单词，比如说：“stock.usd.nyse”, “nyse.vmw”, “quick.orange.rabbit”.这种类型的。当然这个单词列表最多不能超过 255 个字节。

在这个规则列表中，其中有两个替换符是大家需要注意的

*(星号)可以代替一个单词

#(井号)可以替代零个或多个单词

5.6.3. Topic 匹配案例

下图绑定关系如下

Q1–>绑定的是

中间带 orange 带 3 个单词的字符串(.orange.)

Q2–>绑定的是

最后一个单词是 rabbit 的 3 个单词(..rabbit)

第一个单词是 lazy 的多个单词(lazy.#)

上图是一个队列绑定关系图，我们来看看他们之间数据接收情况是怎么样的

quick.orange.rabbit	被队列 Q1Q2 接收到
lazy.orange.elephant	被队列 Q1Q2 接收到
quick.orange.fox	被队列 Q1 接收到
lazy.brown.fox	被队列 Q2 接收到
lazy.pink.rabbit	虽然满足两个绑定但只被队列 Q2 接收一次
quick.brown.fox	不匹配任何绑定不会被任何队列接收到会被丢弃
quick.orange.male.rabbit	是四个单词不匹配任何绑定会被丢弃
lazy.orange.male.rabbit	是四个单词但匹配 Q2

当队列绑定关系是下列这种情况时需要引起注意

当一个队列绑定键是#,那么这个队列将接收所有数据，就有点像 fanout 了

如果队列绑定键当中没有#和*出现，那么该队列绑定类型就是 direct 了

5.6.4. 实战

public class EmitLogTopic {
private static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] argv) throws Exception
{ try (Channel channel = RabbitUtils.getChannel()) { channel.exchangeDeclare(EXCHANGE_NAME, "topic");
/**
*  Q1-->绑定的是
*       中间带 orange 带 3 个单词的字符串(*.orange.*)
*  Q2-->绑定的是
*       最后一个单词是 rabbit 的 3 个单词(*.*.rabbit)
*       第一个单词是 lazy 的多个单词(lazy.#)
*
*/
Map<String, String> bindingKeyMap = new HashMap<>(); 
bindingKeyMap.put("quick.orange.rabbit","被队列 Q1Q2 接收到");
 bindingKeyMap.put("lazy.orange.elephant","被队列 Q1Q2 接收到");
  bindingKeyMap.put("quick.orange.fox","被队列 Q1 接收到"); 
  bindingKeyMap.put("lazy.brown.fox","被队列 Q2 接收到");
bindingKeyMap.put("lazy.pink.rabbit","虽然满足两个绑定但只被队列 Q2 接收一次");
 bindingKeyMap.put("quick.brown.fox","不匹配任何绑定不会被任何队列接收到会被丢弃");
  bindingKeyMap.put("quick.orange.male.rabbit","是四个单词不匹配任何绑定会被丢弃");
   bindingKeyMap.put("lazy.orange.male.rabbit","是四个单词但匹配 Q2");
for (Map.Entry<String, String> bindingKeyEntry: bindingKeyMap.entrySet()){ String bindingKey = bindingKeyEntry.getKey();
String message = bindingKeyEntry.getValue(); channel.basicPublish(EXCHANGE_NAME,bindingKey, null,
message.getBytes("UTF-8"));
System.out.println("生产者发出消息" + message);
}
}
}
}

public class ReceiveLogsTopic01 {
private static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "topic");
//声明 Q1 队列与绑定关系
String queueName="Q1";
channel.queueDeclare(queueName, false, false, false, null); channel.queueBind(queueName, EXCHANGE_NAME, "*.orange.*");
System.out.println("等待接收消息 ........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) ->
{ String message = new String(delivery.getBody(), "UTF-8");
System.out.println("           接           收          队           列          :"+queueName+"                                                        绑           定
键:"+delivery.getEnvelope().getRoutingKey()+",消息:"+message); };
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
}
}

public class ReceiveLogsTopic02 {
private static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "topic");
//声明 Q2 队列与绑定关系
String queueName="Q2";
channel.queueDeclare(queueName, false, false, false, null); channel.queueBind(queueName, EXCHANGE_NAME, "*.*.rabbit"); channel.queueBind(queueName, EXCHANGE_NAME, "lazy.#");
System.out.println("等待接收消息 ........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) ->
{ String message = new String(delivery.getBody(), "UTF-8");
System.out.println("           接           收          队           列          :"+queueName+"                                                        绑           定
键:"+delivery.getEnvelope().getRoutingKey()+",消息:"+message); };
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
}
}

自己写的

package com.atguigu.rabbitmq.server;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;
/**
 * 生产者
 */
public class EmitLogTopic {
    //交换机的名称
    public static final String EXCHANGE_NAME = "topic_logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqutils.getChannel();
        Map<String, String> bindingKeyMap = new HashMap<>();
        bindingKeyMap.put("quick.orange.rabbit", "被队列 Q1Q2 接收到");
        bindingKeyMap.put("lazy.orange.elephant", "被队列 Q1Q2 接收到");
        bindingKeyMap.put("quick.orange.fox", "被队列 Q1 接收到");
        bindingKeyMap.put("lazy.brown.fox", "被队列 Q2 接收到");
        bindingKeyMap.put("lazy.pink.rabbit", "虽然满足两个绑定但只被队列 Q2 接收一次");
        bindingKeyMap.put("quick.brown.fox", "不匹配任何绑定不会被任何队列接收到会被丢弃");
        bindingKeyMap.put("quick.orange.male.rabbit", "是四个单词不匹配任何绑定会被丢弃");
        bindingKeyMap.put("lazy.orange.male.rabbit", "是四个单词但匹配 Q2");
        for (Map.Entry<String, String> bindingKeyEntry : bindingKeyMap.entrySet()) {
            String bindingKey = bindingKeyEntry.getKey();
            String message = bindingKeyEntry.getValue();
            channel.basicPublish(EXCHANGE_NAME, bindingKey, null,
                    message.getBytes("UTF-8"));
            System.out.println("生产者发出消息" + message);
        }
    }
    }
 package com.atguigu.rabbitmq.server;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
import com.sun.corba.se.spi.legacy.interceptor.ORBInitInfoExt;
import java.io.IOException;
import java.net.BindException;
import java.util.concurrent.TimeoutException;
/**
 * 声明主题交换机相关的队列
 *
 * 消费者C1
 */
public class ReceiveLogTopic01 {
    //交换机的名称
    public static final String EXCHANGE_NAME="topic_logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqutils.getChannel();
     //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        //声明队列名字
        String queueName="Q1";
        channel.queueDeclare(queueName,false,false,false,null);
        /**
         *
         * 1.队列名字
         *  2.交换机的名字
         *  3.key的值
         */
        channel.queueBind(queueName,EXCHANGE_NAME,"*.orange.*");
        System.out.println("等待接受消息");
        DeliverCallback deliverCallback= (var1,var2)->{
            System.out.println(new String(var2.getBody(),"utf-8"));
            System.out.println("接受队列"+queueName+"绑定建"+var2.getEnvelope().getRoutingKey());
        };
        //接受消息
        channel.basicConsume(queueName,true,deliverCallback,var1->{});
    }
}
package com.atguigu.rabbitmq.server;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * 声明主题交换机相关的队列
 *
 * 消费者C2
 */
public class ReceiveLogTopic02 {
    //交换机的名称
    public static final String EXCHANGE_NAME="topic_logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqutils.getChannel();
     //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        //声明队列名字
        String queueName="Q2";
        channel.queueDeclare(queueName,false,false,false,null);
        /**
         *
         * 1.队列名字
         *  2.交换机的名字
         *  3.key的值
         */
        channel.queueBind(queueName,EXCHANGE_NAME,"*.*.rabbit");
        channel.queueBind(queueName,EXCHANGE_NAME,"lazy.#");
        System.out.println("等待接受消息");
        DeliverCallback deliverCallback= (var1,var2)->{
            System.out.println(new String(var2.getBody(),"utf-8"));
            System.out.println("接受队列"+queueName+"绑定建"+var2.getEnvelope().getRoutingKey());
        };
        //接受消息
        channel.basicConsume(queueName,true,deliverCallback,var1->{});
    }
}

RibbitMQ学习笔记之交换机实战（二）

5.5.4. 实战

5.6. Topics 主题交换机

5.6.1. 之前类型的问题

5.6.2. Topic 的要求

5.6.4. 实战

热门文章

最新文章

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

RibbitMQ学习笔记之交换机实战（二）

5.5.4. 实战

5.6. Topics 主题交换机

5.6.1. 之前类型的问题

5.6.2. Topic 的要求

5.6.4. 实战

热门文章

最新文章

相关电子书