//生产者
class Program{//实体类public class tblUser{public string Name { get; set; }public int Age { get; set; }public int Sex { get; set; }public string CreateTime { get; set; }public int Disable { get; set; }}public static int Interval = 0;public static int num = 1;private static Thread t1;private static Thread t2;private static Thread t3;private static Thread t4;private static Thread t5;public static ConnectionFactory factory = new ConnectionFactory(){//其他的属性不想写,简洁点 不然可能报错  偷一波懒VirtualHost = "/",//虚拟主机//rabbitmq 账户信息UserName = "wnName",Password = "wnPwd",};static void Main(string[] args){try{//Task t1 = Task.Factory.StartNew(() =>//{//    RunModule();//});///*创建任务 t2  t2 执行 数据集合添加操作*///Task t2 = Task.Factory.StartNew(() =>//{//    RunModule();//});t1 = new Thread(new ThreadStart(RunModule));//无参数的委托t2 = new Thread(new ThreadStart(RunModule));//无参数的委托t3 = new Thread(new ThreadStart(RunModule));//无参数的委托t4 = new Thread(new ThreadStart(RunModule));//无参数的委托t5 = new Thread(new ThreadStart(RunModule));//无参数的委托t1.Start();t2.Start();t3.Start();t4.Start();t5.Start();}catch (Exception ex){throw;}Console.ReadKey();}//队列名public static string mqName = "test-User";public static object obj = new object();static void RunModule(){//压入20条数据 可用 rabbitmqctl list_users 查看while (num <= 20){//锁 防止并发lock (obj){using (var connection = factory.CreateConnection()){using (var channel = connection.CreateModel()){//QueueDeclare方法中的参数  queue 队列   passive被动   durable持久  exclusive专属  autoDelete自动删除  nowait现在channel.QueueDeclare(mqName, false, false, false, null);//实例化实体类赋值然后转成json格式压入队列中。为什么要转成json格式？这个格式随意,队列中保存的都是字符串。var jsonData = JsonConvert.SerializeObject(　　　　　　　　　　　　　　　　　　　　new tblUser() { Name = "数据" + num, Age = 0, Sex = 0, 　　　　　　　　　　　　　　　　　　　　CreateTime = DateTime.Now.ToString("yyyy-MM-dd"), Disable = 1 }); //序列化var properties = channel.CreateBasicProperties();properties.DeliveryMode = 2;//编码下var body = Encoding.UTF8.GetBytes(jsonData);channel.BasicPublish("", mqName, properties, body);}}Console.WriteLine("Starting...测试线程,第" + num+"次执行。");num++;}System.Threading.Thread.Sleep(1000);}//终止线程t1.Abort();t2.Abort();t3.Abort();t4.Abort();t5.Abort();t6.Abort();}}

    class Program{private static Thread get1;//线程一private static int  Interval = 0;private static Thread times;//计时器public  static string mqName = "test-User";public  static ConnectionFactory factory = new ConnectionFactory(){VirtualHost = "/",UserName = "wnName",Password = "wnPwd",};static void Main(string[] args){using (var connection = factory.CreateConnection()){using (var channel = connection.CreateModel()){channel.QueueDeclare(mqName, false, false, false, null);var consumer = new QueueingBasicConsumer(channel);channel.BasicConsume(mqName, true, consumer);while (true){var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();var body = ea.Body;var message = Encoding.UTF8.GetString(body);Console.WriteLine("获得数据 {0}", message);Thread.Sleep(1000);}}}}}