RabbitMQ 笔记-摩杜云开发者社区

管理页面

启动管理插件：用于启动管理API和管理页面等

rabbitmq-plugins enable rabbitmq_management

用户/密码 gust/gust 该用户只能通过localhost访问管理页面

RabbitMQ Management

创建用户权限

# 创建用户 （windows）
rabbitmqctl.bat add_user 'username' '9a55f70a841f18b97c3a7db939b7adc9e34a0f1d'

# create a user 创建用户
rabbitmqctl add_user full_access s3crEt
# tag the user with "administrator" for full management UI and HTTP API access 修改用户权限
rabbitmqctl set_user_tags full_access administrator

用户管理

身份验证、授权、访问控制 — RabbitMQ

# 创建用户
rabbitmqctl add_user huangyan 123

# 删除用户
rabbitmqctl delete_user huangyan

# 虚拟机用户授权
# 第一个 ".*" 配置权限
# 第二个 ".*" 写入权限
# 第三个 ".*" 读取权限
# -p 后面是虚拟机名称 
rabbitmqctl set_permissions -p custom-vhost huangyan ".*" ".*" ".*"
#默认的虚拟机名称是 /
rabbitmqctl set_permissions -p / huangyan ".*" ".*" ".*"
# 撤销授权
rabbitmqctl clear_permissions -p "custom-vhost" "username"

#设置管理用户角色
rabbitmqctl set_user_tags huangyan administrator

#修改用户密码
rabbitmqctl change_password huangyan Newpassword

管理页面角色

Management Plugin — RabbitMQ

Tag	Capabilities
(None)	No access to the management plugin
management	Anything the user could do via messaging protocols plus:List virtual hosts to which they can log in via AMQPView all queues, exchanges and bindings in "their" virtual hostsView and close their own channels and connectionsView "global" statistics covering all their virtual hosts, including activity by other users within them
policymaker	Everything "management" can plus:View, create and delete policies and parameters for virtual hosts to which they can log in via AMQP
monitoring	Everything "management" can plus:List all virtual hosts, including ones they could not access using messaging protocolsView other users's connections and channelsView node-level data such as memory use and clusteringView truly global statistics for all virtual hosts
administrator	Everything "policymaker" and "monitoring" can plus:Create and delete virtual hostsView, create and delete usersView, create and delete permissionsClose other users's connections

连接 URL 说明

RabbitMQ URI Specification — RabbitMQ

URI	Username	Password	Host	Port	Vhost
amqp://user:pass@host:10000/vhost	"user"	"pass"	"host"	10000	"vhost"
amqp://user%61:%61pass@ho%61st:10000/v%2fhost	"usera"	"apass"	"hoast"	10000	"v/host"
amqp://
amqp://:@/	""	""			""
amqp://user@	"user"
amqp://user:pass@	"user"	"pass"
amqp://host			"host"
amqp://:10000				10000
amqp:///vhost					"vhost"
amqp://host/			"host"		""
amqp://host/%2f			"host"		"/"
amqp://[::1]			"[::1]" (i.e. the IPv6 address ::1)

消息确认

消费者确认

//确认
channel.ack(msg);

//拒绝
channel.nack(msg);

错误手动确认

被遗忘的确认

错过ack是一个常见的错误。这是一个容易犯的错误，但后果是严重的。当您的客户端退出时，消息将被重新传递（这可能看起来像随机重新传递），但RabbitMQ将消耗越来越多的内存，因为它将无法释放任何未添加的消息。

为了调试这种错误，您可以使用rabbitmqctl打印messages_unacknowledged字段：

sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

在 Windows 上，删除 sudo：

rabbitmqctl.bat list_queues name messages_ready messages_unacknowledged

持久化

声明持久化队列

channel.assertQueue('hello', {durable: true});

声明消息持久化

channel.sendToQueue(queue, Buffer.from(msg), {persistent: true});

公平调度

这告诉 RabbitMQ 不要一次向一个工作人员发送多条消息。或者，换句话说，在处理并确认前一个消息之前，不要向工作人员发送新消息。相反，它会将其调度给下一个尚未繁忙的工作人员。

channel.prefetch(1);

例子

生产者

var amqp = require('amqplib/callback_api');

//连接
amqp.connect('amqp://huangyan:123@172.19.37.45:5672', function(error0, connection) {
  if (error0) {
    throw error0;
  }
    
  //创建管道
  connection.createChannel(function(error1, channel) {
    if (error1) {
      throw error1;
    }
    var queue = 'task_queue';
    var msg = process.argv.slice(2).join(' ') || "Hello World!";

    channel.assertQueue(queue, {
      //持久化队列
      durable: true
    });
      
    //发送消息
    channel.sendToQueue(queue, Buffer.from(msg), {
      //持久化消息
      persistent: true
    });
    console.log(" [x] Sent '%s'", msg);
  });
    
  setTimeout(function() {
    connection.close();
    process.exit(0)
  }, 500);
});

消费者

var amqp = require('amqplib/callback_api');

amqp.connect('amqp://localhost', function(error0, connection) {
  if (error0) {
    throw error0;
  }
  connection.createChannel(function(error1, channel) {
    if (error1) {
      throw error1;
    }
    var queue = 'task_queue';

    channel.assertQueue(queue, {
      durable: true
    });
      
    //标记为预读取，用于rabbitmq调度
    channel.prefetch(1);
    console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue);
    //消耗消息
    channel.consume(queue, function(msg) {
      var secs = msg.content.toString().split('.').length - 1;

      console.log(" [x] Received %s", msg.content.toString());
      setTimeout(function() {
        console.log(" [x] Done");
        //确认消息
        channel.ack(msg);
      }, secs * 1000);
    }, {
      // manual acknowledgment mode,
      // see ../confirms.html for details
      //确认模式为手动
      noAck: false
    });
  });
});