c#生产/消费RabbitMQ

public sealed class JsonSerializer

    {

        public static byte[] Serialize(object message)

        {

            return Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message));

        }

        public static object Deserialize<T>(byte[] bytes)

        {

            return JsonConvert.DeserializeObject<T>(Encoding.UTF8.GetString(bytes));

        }

    }

    public sealed class BinarySerializer

    {

        public static byte[] SerializeToBytes(object obj)

        {

            var formatter = new BinaryFormatter();

            using (var stream = new MemoryStream())

            {

                formatter.Serialize(stream, obj);

                return StreamUtil.ReadAllBytes(stream);

            }

        }

        public static object DeserializeFromBytes(byte[] bytes)

        {

            var formatter = new BinaryFormatter();

            using (var stream = new MemoryStream(bytes))

            {

                return formatter.Deserialize(stream);

            }

        }

    }

private static bool RawPublishMessage()

        {

            var exchange = "TestExchangeRouting...";

            var routingKey = "rk";

            Uri uri = new Uri("amqp://192.168.1.1:1234/");

            ConnectionFactory factory = new ConnectionFactory();

            factory.UserName = "guest";

            factory.Password = "guest";

            factory.VirtualHost = "/";

            factory.RequestedHeartbeat = ;

            factory.Endpoint = new AmqpTcpEndpoint(uri);

            //创建一个连接

            using (IConnection connection = factory.CreateConnection())

            {

                //创建一个通道

                using (IModel channel = connection.CreateModel())

                {

                    //声明一个路由

                    channel.ExchangeDeclare(exchange, "direct");

                    var queueOk = channel.QueueDeclare("testQueue", true, false, false, null);

                    channel.QueueBind(queueOk.QueueName, exchange, routingKey);

                    var model = new Order

                    {

                        Id = ,

                        Title = "工一一个测试Test"

                    }; //这个才是具体的发送内容  

                    var body = JsonSerializer.Serialize(model);

                    var properties = channel.CreateBasicProperties();

                    properties.SetPersistent(true);

                    properties.ContentType = typeof(Order).AssemblyQualifiedName;

                    properties.ContentEncoding = "JSON";

                    //写入

                    channel.BasicPublish(exchange, routingKey, properties, body);

                    Console.WriteLine("写入成功");

                }

            }

            return false;

        }

        private static bool RawGetMessage()

        {

            var exchange = "TestExchangeRouting...";

            var routingKey = "rk";

            Uri uri = new Uri("amqp://192.168.1.1:1234/");

            ConnectionFactory factory = new ConnectionFactory();

            factory.UserName = "guest";

            factory.Password = "guest";

            factory.VirtualHost = "/";

            factory.RequestedHeartbeat = ;

            factory.Endpoint = new AmqpTcpEndpoint(uri);

            //创建一个连接

            using (IConnection connection = factory.CreateConnection())

            {

                //创建一个通道

                using (IModel channel = connection.CreateModel())

                {

                    var basicConsumer = new QueueingBasicConsumer(channel);

                    channel.BasicConsume("testQueue", false, basicConsumer);

                    while (true)

                    {

                        try

                        {

                            BasicDeliverEventArgs basicDeliverEventArgs  ;

                            basicConsumer.Queue.Dequeue(, out basicDeliverEventArgs);

                            if (basicDeliverEventArgs == null)

                            {

                                break;

                            }

                            Task.Run(() =>

                            {

                                channel.BasicAck(basicDeliverEventArgs.DeliveryTag, false);

                            }).Wait();

                            var body = JsonSerializer.Deserialize<Order>(basicDeliverEventArgs.Body);

                            Console.WriteLine(string.Format("RoutingKey:{0},Body:{1}", basicDeliverEventArgs.RoutingKey,

                                 JsonConvert.SerializeObject(body, Formatting.Indented)));

                        }

                        catch (Exception)

                        {

                            break;

                        }

                    }

                    channel.Close();

                }

                connection.Close();

            }

            return false;

        }

public static byte[] ReadAllBytes(Stream stream)

        {           

            var bytes = new byte[stream.Length];

            stream.Seek(, SeekOrigin.Begin);

            for (var i = ; i < stream.Length; i++)

            {

                bytes[i] = (byte)stream.ReadByte();

            }

            return bytes;

        }