当前位置：首页 > wzjs >正文

重庆做网站微信的公司网站建设推销拜访客户怎么开头

wzjs 2025/9/6 0:23:28

重庆做网站微信的公司,网站建设推销拜访客户怎么开头,wordpress默认字体大小,怎么做软件系统开发Topics(通配符模式) Topics 和Routing模式的区别是: topics 模式使⽤的交换机类型为topic(Routing模式使⽤的交换机类型为direct)topic 类型的交换机在匹配规则上进⾏了扩展, Binding Key⽀持通配符匹配(direct类型的交换机路由规则是BindingKey和RoutingKey完全匹配) 在top…

Topics(通配符模式)

Topics 和Routing模式的区别是:

topics 模式使⽤的交换机类型为topic(Routing模式使⽤的交换机类型为direct)
topic 类型的交换机在匹配规则上进⾏了扩展, Binding Key⽀持通配符匹配(direct类型的交换机路由规则是BindingKey和RoutingKey完全匹配)

在topic类型的交换机在匹配规则上, 有些要求:

RoutingKey 是⼀系列由点( . )分隔的单词, ⽐如 " stock.usd.nyse ", " nyse.vmw ", " quick.orange.rabbit "
BindingKey 和RoutingKey⼀样, 也是点( . )分割的字符串
Binding Key中可以存在两种特殊字符串, ⽤于模糊匹配

* 表⽰⼀个单词
# 表⽰多个单词(0-N个)

⽐如:

Binding Key 为"d.a.b" 会同时路由到Q1 和Q2
Binding Key 为"d.a.f" 会路由到Q1
Binding Key 为"c.e.f" 会路由到Q2
Binding Key 为"d.b.f" 会被丢弃, 或者返回给⽣产者(需要设置mandatory参数)

引⼊依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version>
</dependency>

编写⽣产者代码和路由模式, 发布订阅模式的区别是:

交换机类型不同, 绑定队列的RoutingKey不同

创建交换机

定义交换机类型为BuiltinExchangeType.TOPIC

channel.exchangeDeclare(Constants.TOPIC_EXCHANGE_NAME,
BuiltinExchangeType.TOPIC, true, false, false, null);

声明队列

channel.queueDeclare(Constants.TOPIC_QUEUE_NAME1, true, false, false, null);
channel.queueDeclare(Constants.TOPIC_QUEUE_NAME2, true, false, false, null);

绑定交换机和队列

//队列1绑定error, 仅接收error信息
channel.queueBind(Constants.TOPIC_QUEUE_NAME1,Constants.TOPIC_EXCHANGE_NAME,
"*.error");
//队列2绑定info, error: error,info信息都接收
channel.queueBind(Constants.TOPIC_QUEUE_NAME2,Constants.TOPIC_EXCHANGE_NAME,
"#.info");
channel.queueBind(Constants.TOPIC_QUEUE_NAME2,Constants.TOPIC_EXCHANGE_NAME,
"*.error");

发送消息

String msg = "hello topic, I'm order.error";
channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"order.error",null,msg.getBy
tes());
String msg_black = "hello topic, I'm order.pay.info";
channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"order.pay.info",null,msg_bl
ack.getBytes());
String msg_green= "hello topic, I'm pay.error";
channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"pay.error",null,msg_green.g
etBytes());

完整代码:

public static String TOPIC_EXCHANGE_NAME = "test_topic";
public static String TOPIC_QUEUE_NAME1 = "topic_queue1";
public static String TOPIC_QUEUE_NAME2 = "topic_queue2";import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import constant.Constants;
public class TopicRabbitProducer {public static void main(String[] args) throws Exception {//1. 创建channel通道ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);//ip 默认值localhostfactory.setPort(Constants.PORT); //默认值5672factory.setVirtualHost(Constants.VIRTUAL_HOST);//虚拟机名称, 默认 /factory.setUsername(Constants.USER_NAME);//⽤⼾名,默认guestfactory.setPassword(Constants.PASSWORD);//密码, 默认guestConnection connection = factory.newConnection();Channel channel = connection.createChannel();//2. 创建交换机channel.exchangeDeclare(Constants.TOPIC_EXCHANGE_NAME,
BuiltinExchangeType.TOPIC, true, false, false, null);//3. 声明队列//如果没有⼀个这样的⼀个队列, 会⾃动创建, 如果有, 则不创建channel.queueDeclare(Constants.TOPIC_QUEUE_NAME1, true, false, false,
null);channel.queueDeclare(Constants.TOPIC_QUEUE_NAME2, true, false, false,
null);//4. 绑定队列和交换机//队列1绑定error, 仅接收error信息channel.queueBind(Constants.TOPIC_QUEUE_NAME1,Constants.TOPIC_EXCHANGE_NAME,
"*.error");//队列2绑定info, error: error,info信息都接收channel.queueBind(Constants.TOPIC_QUEUE_NAME2,Constants.TOPIC_EXCHANGE_NAME,
"#.info");channel.queueBind(Constants.TOPIC_QUEUE_NAME2,Constants.TOPIC_EXCHANGE_NAME,
"*.error");//5. 发送消息String msg = "hello topic, I'm order.error";channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"order.error",null,msg.getBy
tes());String msg_black = "hello topic, I'm order.pay.info";channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"order.pay.info",null,msg_bl
ack.getBytes());String msg_green= "hello topic, I'm pay.error";channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"pay.error",null,msg_green.g
etBytes());//6.释放资源channel.close();connection.close();}
}

编写消费者代码

Routing模式的消费者代码和Routing模式代码⼀样, 修改消费的队列名称即可

同样复制出来两份

消费者1:TopicRabbitmqConsumer1

消费者2: TopicRabbitmqConsumer2

完整代码:

import com.rabbitmq.client.*;
import constant.Constants;
import java.io.IOException;
public class TopicRabbitmqConsumer1 {public static void main(String[] args) throws Exception {//1. 创建channel通道ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);//ip 默认值localhostfactory.setPort(Constants.PORT); //默认值5672factory.setVirtualHost(Constants.VIRTUAL_HOST);//虚拟机名称, 默认 /factory.setUsername(Constants.USER_NAME);//⽤⼾名,默认guestfactory.setPassword(Constants.PASSWORD);//密码, 默认guestConnection connection = factory.newConnection();Channel channel = connection.createChannel();//2. 接收消息, 并消费DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息: " + new String(body));}};channel.basicConsume(Constants.TOPIC_QUEUE_NAME1, true, consumer);}
}

运⾏程序, 观察结果

运⾏⽣产者, 可以看到队列的消息数

运⾏消费者

RPC(RPC通信)

RPC(Remote Procedure Call), 即远程过程调⽤. 它是⼀种通过⽹络从远程计算机上请求服务, ⽽不需要了解底层⽹络的技术. 类似于Http远程调⽤

RabbitMQ实现RPC通信的过程, ⼤概是通过两个队列实现⼀个可回调的过程

⼤概流程如下:

客⼾端发送消息到⼀个指定的队列, 并在消息属性中设置replyTo字段, 这个字段指定了⼀个回调队列, 服务端处理后, 会把响应结果发送到这个队列
服务端接收到请求后, 处理请求并发送响应消息到replyTo指定的回调队列
客⼾端在回调队列上等待响应消息. ⼀旦收到响应，客⼾端会检查消息的correlationId属性，以确保它是所期望的响应

引⼊依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version>
</dependency>

编写客⼾端代码客⼾端代码主要流程如下:

声明两个队列, 包含回调队列replyQueueName, 声明本次请求的唯⼀标志corrId
将replyQueueName和corrId配置到要发送的消息队列中
使⽤阻塞队列来阻塞当前进程, 监听回调队列中的消息, 把请求放到阻塞队列中
阻塞队列有消息后, 主线程被唤醒，打印返回内容

声明队列

//2. 声明队列, 发送消息
channel.queueDeclare(Constants.RPC_REQUEST_QUEUE_NAME, true, false, false,
null);

定义回调队列

// 定义临时队列，并返回⽣成的队列名称
String replyQueueName = channel.queueDeclare().getQueue();

使⽤内置交换机发送消息

// 本次请求唯⼀标志
String corrId = UUID.randomUUID().toString();
// ⽣成发送消息的属性
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId) // 唯⼀标志本次请求.replyTo(replyQueueName) // 设置回调队列.build();
// 通过内置交换机, 发送消息
String message = "hello rpc...";
channel.basicPublish("", Constants.RPC_REQUEST_QUEUE_NAME, props,
message.getBytes());

使⽤阻塞队列, 来存储回调结果

// 阻塞队列，⽤于存储回调结果
final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
//接收服务端的响应
DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到回调消息:"+ new String(body));//如果唯⼀标识正确, 放到阻塞队列中if (properties.getCorrelationId().equals(corrId)) {response.offer(new String(body, "UTF-8"));}}
};
channel.basicConsume(replyQueueName, true, consumer);

获取回调结果

// 获取回调的结果
String result = response.take();
System.out.println(" [RPCClient] Result:" + result);

完整代码

public static String RPC_REQUEST_QUEUE_NAME = "rpc_request_queue";import com.rabbitmq.client.*;
import constant.Constants;
import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class RPCClient {public static void main(String[] args) throws Exception {//1. 创建Channel通道ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);//ip 默认值localhostfactory.setPort(Constants.PORT); //默认值5672factory.setVirtualHost(Constants.VIRTUAL_HOST);//虚拟机名称, 默认 /factory.setUsername(Constants.USER_NAME);//⽤⼾名,默认guestfactory.setPassword(Constants.PASSWORD);//密码, 默认guestConnection connection = factory.newConnection();Channel channel = connection.createChannel();//2. 声明队列channel.queueDeclare(Constants.RPC_REQUEST_QUEUE_NAME, true, false,
false, null);// 唯⼀标志本次请求String corrId = UUID.randomUUID().toString();// 定义临时队列，并返回⽣成的队列名称String replyQueueName = channel.queueDeclare().getQueue();// ⽣成发送消息的属性AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId) // 唯⼀标志本次请求.replyTo(replyQueueName) // 设置回调队列.build();// 通过内置交换机, 发送消息String message = "hello rpc...";channel.basicPublish("", Constants.RPC_REQUEST_QUEUE_NAME, props,
message.getBytes());// 阻塞队列，⽤于存储回调结果final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);//接收服务端的响应DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到回调消息:"+ new String(body));if (properties.getCorrelationId().equals(corrId)) {response.offer(new String(body, "UTF-8"));}}};channel.basicConsume(replyQueueName, true, consumer);// 获取回调的结果String result = response.take();System.out.println(" [RPCClient] Result:" + result);//释放资源channel.close();connection.close();}
}