RabbitMQ实现路由模式发送接收消息
1.基础类
public class RabbitMQHelper
{
public static IConnection GetConnect()
{
//连接工厂
var fac = new ConnectionFactory()
{
HostName = “127.0.0.1”,//IP
Port = 5672,//端口
UserName = “xxx”,
Password = “xxx”,
VirtualHost = “/”
};
return fac.CreateConnection();
}
}
2.发送端
public void SendRoutingnMsg()
{
using (var connection = RabbitMQHelper.GetConnect())
{
using (var channel = connection.CreateModel())
{
//声明交换机对象
string exchangeName = “RoutingExchange”;//交换机名称
channel.ExchangeDeclare(exchangeName, “direct”);
//创建队列
string queueName1 = “RoutingQueue1”;
channel.QueueDeclare(queueName1, false, false, false, null);
string queueName2 = “RoutingQueue2”;
channel.QueueDeclare(queueName2, false, false, false, null);
string queueName3 = “RoutingQueue3”;
channel.QueueDeclare(queueName3, false, false, false, null);
//绑定到交互机
//PubSubExchange 绑定了3个队列
channel.QueueBind(queue: queueName1, exchange: exchangeName, routingKey: “RouteKey1”);
channel.QueueBind(queue: queueName2, exchange: exchangeName, routingKey: “RouteKey2”);
channel.QueueBind(queue: queueName3, exchange: exchangeName, routingKey: “RouteKey3”);
int i = 0;
while (i < 10)
{
string message =
"
R
a
b
b
i
t
M
Q
给
R
o
u
t
e
K
e
y
1
发送的第
i
条消息
"
;
v
a
r
b
o
d
y
=
E
n
c
o
d
i
n
g
.
U
T
F
8.
G
e
t
B
y
t
e
s
(
m
e
s
s
a
g
e
)
;
c
h
a
n
n
e
l
.
B
a
s
i
c
P
u
b
l
i
s
h
(
e
x
c
h
a
n
g
e
:
e
x
c
h
a
n
g
e
N
a
m
e
,
r
o
u
t
i
n
g
K
e
y
:
"
R
o
u
t
e
K
e
y
1
"
,
n
u
l
l
,
b
o
d
y
)
;
C
o
n
s
o
l
e
.
W
r
i
t
e
L
i
n
e
(
"RabbitMQ给RouteKey1发送的第{i}条消息"; var body = Encoding.UTF8.GetBytes(message); channel.BasicPublish(exchange: exchangeName, routingKey: "RouteKey1", null, body); Console.WriteLine(
"RabbitMQ给RouteKey1发送的第i条消息";varbody=Encoding.UTF8.GetBytes(message);channel.BasicPublish(exchange:exchangeName,routingKey:"RouteKey1",null,body);Console.WriteLine(“给RouteKey1,第{i}条Routing消息发送成功！”);
i++;
}
}
}
}
3.接收端
public void ReceivePubSubMsg()
{
var connection = RabbitMQHelper.GetConnect();
{
var channel = connection.CreateModel();
{
//中明exchange
string exchangeName = “RoutingExchange”;
//交换机名称
channel.ExchangeDeclare(exchange: exchangeName, type: “direct”);
//创建队列
string queueNamel = “RoutingQueue1”;
channel.QueueDeclare(queueNamel, false, false, false, null);
//绑定到交互机
channel.QueueBind(queue: queueNamel, exchange: exchangeName, routingKey: “RouteKey1”);
//申明consumer
var consumer = new EventingBasicConsumer(channel);
//绑定消息接收后的事件委托
consumer.Received += (m, e) =>
{
var body = e.Body;
var message = Encoding.UTF8.GetString(body.ToArray());
Console.WriteLine($“接收的消息{message}”, message);
};
channel.BasicConsume(queue: queueNamel, autoAck: true, consumer: consumer);
}
}
}