.Net下RabbitMQ的使用(2) — 发送接收消息

在安装和配置完成RabbitMQ之后，我们就可以尝试做一个最简单的例子：发送和接收消息。

我们先来看客户端也就是发送者的代码：

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
51
52
53
54
55
56
1public class RabbitClient

2

3{

4

5    //定义连接工厂

6

7    ConnectionFactory factory = new ConnectionFactory();

8

9 

10

11    public RabbitClient()

12

13    {

14

15        //指定要连接的RabbitMQ服务地址

16

17        factory.HostName = "localhost";

18

19    }

20

21 

22    public void Send()

23    {

24

25        //定义要发送的数据

26

27        RequestMessage message = new RequestMessage() { MessageId = Guid.NewGuid(), Message = "this is a 请求。" };

28

29        //创建一个 AMQP 连接

30        using (IConnection connection = factory.CreateConnection())

31        {

32

33            using (IModel channel = connection.CreateModel())

34            {

35                //在MQ上定义一个队列

36                channel.QueueDeclare("esbtest.rmq.consoleserver", false, false, false, null);

37

38                //序列化消息对象，RabbitMQ并不支持复杂对象的序列化，所以对于自定义的类型需要自己序列化

39

40                XmlSerializer xs = new XmlSerializer(typeof(RequestMessage));

41                using (MemoryStream ms = new MemoryStream())

42                {

43                    xs.Serialize(ms, message);

44                    byte[] bytes = ms.ToArray();

45                    //指定发送的路由，通过默认的exchange直接发送到指定的队列中。

46                    channel.BasicPublish("", "esbtest.rmq.consoleserver", null, bytes);

47                }

48

49                Console.WriteLine(string.Format("Request Message Sent, Id:{0}, Message:{1}", message.MessageId, message.Message));

50

51            }

52        }

53

54    }

55}

56

在方法

1
2
3
1channel.BasicPublish("", "esbtest.rmq.consoleserver", null, bytes);  

2

3

中的第一个参数是需要输入一个exchange。在RabbitMQ中，所有的消息都必须要通过exchange发送到各个queue里面去。发送者发送消息，其实也就是把消息放到exchange中去。而exchange知道应该把消息放到哪里去。在这个方法中，我们没有输入exchange的名称，只是定义了一个空的echange，而在第二个参数routeKey中输入了我们目标队列的名称。RabbitMQ会帮我定义一个默认的exchange，这个exchange会把消息直接投递到

我们输入的队列中，这样服务端只需要直接去这个定义了的队列中获取消息就可以了。

服务端的代码：

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
1public class RabbitServer

2{

3    ConnectionFactory factory = null;

4

5    public void Listen()

6    {

7        factory = new ConnectionFactory();

8        factory.HostName = "localhost";

9        using (IConnection connection = factory.CreateConnection())

10        {

11            using (IModel channel = connection.CreateModel())

12            {

13                //在MQ上定义一个队列，如果名称相同不会重复创建

14                channel.QueueDeclare("esbtest.rmq.consoleserver", false, false, false, null);

15                Console.WriteLine("Listening...");

16                //在队列上定义一个消费者

17                QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);

18                channel.BasicConsume("esbtest.rmq.consoleserver", true, consumer);

19                while (true)

20                {

21                    //阻塞函数，获取队列中的消息

22                    BasicDeliverEventArgs ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

23

24                    byte[] body = ea.Body;

25                    XmlSerializer xs = new XmlSerializer(typeof(RequestMessage));

26                    using (MemoryStream ms = new MemoryStream(body))

27                    {

28                        RequestMessage message = (RequestMessage)xs.Deserialize(ms);

29                        Console.WriteLine("Receive a Message, Id:" + message.MessageId + " Message:" + message.Message);

30                    }

31                }

32            }

33        }

34    }

35}

36

至此，简单的发送接收程序就可以运行了，运行RabbitMQ，然后分别运行客户端和服务端。运行结果如图：

客户端：

服务端：

代码可以从这里下载。