发布消息：（生产者）
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
/// <summary> /// 发送消息 /// </summary> /// <param name="queue">队列名</param> /// <param name="message">消息内容</param> private static void PublishInfo(string queue, string message) { try { var factory = new ConnectionFactory(); factory.HostName = "localhost";//RabbitMQ服务在本地运行 factory.UserName = "guest";//用户名 factory.Password = "guest";//密码 using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { bool durable = true;//将消息标记为持久性 channel.QueueDeclare(queue, durable, false, false, null); IBasicProperties properties = channel.CreateBasicProperties(); properties.DeliveryMode = 2; var body = Encoding.UTF8.GetBytes(message); channel.BasicPublish("", queue, properties, body); //开始传递 } } } catch (System.Exception ex) { Console.WriteLine(ex.ToString()); } Console.ReadLine(); } |
使用：
|
1
|
PublishInfo("test", "222222222222"); |
点击test进入队列内部：
消费消息：
方法一：旧版可用
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
|
private static void ConsumeInfo(string queue) { try { var factory = new ConnectionFactory(); factory.HostName = "localhost";//RabbitMQ服务在本地运行 factory.UserName = "guest";//用户名 factory.Password = "guest";//密码 using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { //bool durable = true;//将消息标记为持久性 //channel.QueueDeclare("sljcgx:finish-queue", durable, false, false, null);//声明队列，队列不存在则创建该队列 //设置prefetchCount : 1来告知RabbitMQ，在未收到消费端的消息确认时，不再分发消息，也就确保了当消费端处于忙碌状态时，不再分配任务。 channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false); var consumer = new QueueingBasicConsumer(channel); //false为手动确认，获取消息后需要添加 channel.BasicAck(ea.DeliveryTag, false)确认 //true则为自动确认，不需要这行代码 channel.BasicConsume(queue, false, consumer); while (true) { try { var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//阻塞函数，获取队列中的消息 var bodyReceive = ea.Body; string message = Encoding.UTF8.GetString(bodyReceive); Console.WriteLine(message); channel.BasicAck(ea.DeliveryTag, false);// 消息确认信号，确认后该消息将从队列里移除 } catch (System.Exception ex) { Console.WriteLine(ex.Message); } } } } } catch (System.Exception ex) { Console.WriteLine(ex.ToString()); } Console.ReadLine(); } |
方法二：（通用）
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
private static void ConsumeInfo2(string queue) { try { var factory = new ConnectionFactory(); factory.HostName = "localhost";//RabbitMQ服务在本地运行 factory.UserName = "guest";//用户名 factory.Password = "guest";//密码 var connection = factory.CreateConnection(); var channel = connection.CreateModel(); channel.QueueDeclare(queue, true, false, false, null);//声明队列，队列不存在则创建该队列 channel.BasicQos(0, 1, false);//公平分发、同一时间只处理一个消息。 var consumer = new EventingBasicConsumer(channel); //消费者（指定消息通道）该事件在接收到消息时触发 consumer.Received += (sender, e) => { byte[] body = e.Body.ToArray(); //消息字节数组 string message = Encoding.UTF8.GetString(body); //消息内容 Console.WriteLine(message); channel.BasicAck(e.DeliveryTag, false); //手工确认 }; channel.BasicConsume(queue, false, consumer); //消费消息（在当前通道中监听 queue 队列，并进行消费） Console.ReadLine(); connection.Close(); channel.Close(); } catch (System.Exception ex) { Console.WriteLine(ex.ToString()); } } |
到此这篇关于c# rabbitmq 简单收发消息的示例代码的文章就介绍到这了,更多相关c# rabbitmq 收发消息内容请搜索服务器之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持服务器之家！
原文链接：https://blog.csdn.net/fangyu723/article/details/107908066
