rabbitmq 示例

【实例简介】

C# 调用rabbitmq

【核心代码】

 string projectId = "Alrm";

                //服务器所在的主机ip

                //是否对消息队列持久化保存
                bool persistMode = true;
                ConnectionFactory cf = new ConnectionFactory();
                // cf.HostName =  ConstData.UriMq; 
                cf.UserName = ConstData.userName;//某个vhost下的用户
                cf.Password = ConstData.pass;
                // cf.VirtualHost = "gyg001";//vhost
                // cf.RequestedHeartbeat = 0; 
                //cf.Endpoint = new AmqpTcpEndpoint(ConstData.UriMq);
                //  cf.HostName = "localhost";
                cf.Endpoint = new AmqpTcpEndpoint(new Uri("amqp://192.168.8.235:5672/"));
                //创建一个连接到具体总结点的连接
                using (IConnection conn = cf.CreateConnection())
                {             //创建并返回一个新连接到具体节点的通道
                    using (IModel ch = conn.CreateModel())
                    {


                        if (!string.IsNullOrEmpty(uuid))
                        {
                          //  ch.ExchangeDeleteNoWait(projectId,true);
                            ch.QueueDeleteNoWait(uuid,true,true);
                            //声明一个路由
                            ch.ExchangeDeclare(projectId, "topic");
                            //声明一个队列
//                            方法的第四的参数autoDelete，一般都会输入false。文档描述这个参数如果是true的话，
                            //意思是：如果这个queue不再使用（没有被订阅）的话，server就会删除它。在我的测试过程中，
                            //只要是连接改queue的所有接收者都断开连接的话，该queue就会被删除，即使里面还有没有处理的消息。
                            //RabbitMQ的重启也同样会删除他们。如果输入的是false，那与之相连的客户端都断开连接的话，
                            //服务是不会删除这个队列的，队列中的消息也就会存在。
                            
                            //发送端在没有客户端连接的时候也可以把消息放入改队列，客户端起来的时候，就会得到这些消息。
                            //但是如果RabbitMQ服务重启的话，该队列就没有了，里面的消息自然也就没有了。

                            //第三个参数是exclusive，文档描述说，如果是true，那么申明这个queue的connection断了，
                            //那么这个队列就被删除了，包括里面的消息。

                             //第二个参数durable，文档描述说，如果是true，则代表是一个持久的队列，那么在服务重启后，也会存在。
                            //因为服务会把持久化的queue存放在硬盘上，放服务重启的时候，会重新申明这个queue。
                            //当然必须是在autoDelete和exclusive都为false的时候。队列是可以被持久化，
                            //但是里面的消息是否为持久化那还要看消息的持久化设置。
                            //也就是说，如果重启之前那个queue里面还有没有发出去的消息的话，
                            //重启之后那队列里面是不是还存在原来的消息，
                            //这个就要取决于发送者在发送消息时对消息的设置了。

                            ////信息过期时限
                            IDictionary<String, Object> args = new Dictionary<String, Object>();
                           // args.Add("x-message-ttl", 30000);
                          
                            ///队列过期时间
                            args.Add("x-expires", 120000);


                                ch.QueueDeclare(uuid, true, false, false, args);
                                //将一个队列和一个路由绑定起来。并制定路由关键字  
                                ///uuid binding key  
                                ch.QueueBind(uuid, projectId, "User.*");
                                ch.BasicQos(0,1,false);
                               
                          
                           
                        }

 var consumer = new EventingBasicConsumer(ch);
                            consumer.Received = (model, ea) =>
                            {
                                try{
                                int n = r.Next(5);

                                var body = ea.Body;
                                var message = Encoding.UTF8.GetString(body);
                                Console.WriteLine("  Received {0} sleep={1}*100ms -- {2}", message, n, DateTime.Now.Millisecond);
                              //  Thread.Sleep(n * 100);
                                if (n % 5 == 3)
                                {
                                    Console.WriteLine("  Received {0} 触发发送失败事件 消息不处理重回队列 {1})", message,DateTime.Now.Millisecond);
                                   // ch.BasicAck(deliveryTag: ea.DeliveryTag, multiple: true); 
                                    ch.BasicNack(ea.DeliveryTag,true,true);
                                }
                                else {
                                    ch.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                                }
                                }
                                catch (Exception ee) {
                                
                                }
                            };
                            ch.BasicConsume(queue: "User0", noAck: false, consumer: consumer);
                            Console.ReadKey();