.net core RabbitMQ 消息队列

上篇我们说到erlang的安装,现在有了基础前提,就可以继续安装RabbitMQ了!
这里我选用的RabbitMQ版本是:

PS:这个RabbitMQ版本是要对应前面erlang版本,所以前面我们安装的版本是21.2,因为需要提供21.X版本的erlang才能安装 rabbitmq-server-3.7.10

1.安装RabbitMQ
下载地址:http://www.rabbitmq.com/install-windows.html
download完,执行exe文件,安装到自己选用的目录,并配置环境变量

rabbitmq的基本操作:

  • 启动:rabbitmq-server -detached
  • 关闭:rabbitmqctl stop
  • 启动:rabbitmqctl status

2.配置rabbitmq网页管理插件
以管理员运行命令提示启用插件:

rabbitmq-plugins enable rabbitmq_management

打开浏览器页面:http://localhost:15672 可以看到
.net core RabbitMQ 消息队列

默认登陆为:guest/guest

3.开启rabbitMQ远程访问

  • 添加用户,用户名:XRom 密码:XRom123
rabbitmqctl add_user XRom XRom123
  • 添加权限
rabbitmqctl set_permissions -p "/" XRom ".*" ".*" ".*"
  • 修改用户角色
rabbitmqctl set_user_tags XRom administrator

然后就可以远程访问了,可以用新增的用户登录RabbitMQ

4.Producer与Exchange

  • Producer
    消息的生产者,也就是创建消息的对象
  • Exchange
    消息的接受者,也就是用来接收消息的对象,Exchange接收到消息后将消息按照规则发送到与他绑定的Queue中。下面我们来定义一个Producer与Exchange。

新建.net core 控制台项目,并引入NuGet包

接下来可以用代码看实现效果:

using RabbitMQ.Client;using RabbitMQ.Client.Events;using System;using System.Text;using System.Threading;namespace RabbitMQConsole{ class Program { /// <summary> /// 创建只读连接对象 /// </summary> private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory() { HostName = "",//这里写自己电脑hostname,可以通过命令提示符,直接输入hostname查询 Port = 5672, UserName = "XRom", Password = "XRom123", VirtualHost = "/" }; static void Main(string[] args) { var exchange = "change2"; var route = "route2"; var queue = "queue2"; using (IConnection conn = rabbitMqFactory.CreateConnection()) { using (IModel channel = conn.CreateModel()) { channel.ExchangeDeclare(exchange, type: "direct", durable: true, autoDelete: false);//创建change2 channel.QueueDeclare(queue, durable: true, exclusive: false, autoDelete: false);//创建queue2 channel.QueueBind(queue, exchange, route);//将queue2绑定到change2 #region 发送消息 var props = channel.CreateBasicProperties(); props.Persistent = true; //持久化 channel.BasicPublish(exchange, route, true, props, Encoding.UTF8.GetBytes("hello rabbitmq!!")); #endregion #region 消费消息 //while (true) //{ // var message = channel.BasicGet(queue, true); //第二个参数说明自动释放消息,如为false需手动释放消息 // if (message != null) // { // var msgBody = Encoding.UTF8.GetString(message.Body); // Console.WriteLine(string.Format("***接收时间:{0},消息内容:{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody)); // } // System.Threading.Thread.Sleep(TimeSpan.FromSeconds(1)); //} #endregion #region 让失败的消息回到队列中 //while (true) //{ // var message = channel.BasicGet(queue, false); // if (message != null) // { // var msgBody = Encoding.UTF8.GetString(message.Body); // Console.WriteLine(string.Format("***接收时间:{0},消息内容:{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody)); // Console.WriteLine(message.DeliveryTag);   //当前消息被处理的次序数 // if (1 == 1) // channel.BasicReject(message.DeliveryTag, true); // } // System.Threading.Thread.Sleep(TimeSpan.FromSeconds(1)); //} #endregion #region 监听消息 //channel.BasicQos(prefetchSize: 0, prefetchCount: 20, global: false); //一次接受10条消息,否则rabbit会把所有的消息一次性推到client,会增大client的负荷 //EventingBasicConsumer consumer = new EventingBasicConsumer(channel); //consumer.Received += (model, ea) => //{ // Byte[] body = ea.Body; // String message = Encoding.UTF8.GetString(body); // Console.WriteLine(message + Thread.CurrentThread.ManagedThreadId); // channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); //}; //channel.BasicConsume(queue: queue, autoAck: false, consumer: consumer); //Console.ReadLine(); #endregion } } } }}


相关文章