systemctl enable rabbitmq-server

systemctl start rabbitmq-server

systemctl stop rabbitmq-server

less /var/log/rabbitmq/rabbit@localhost.log

基本配置

RabbitMQ有一套默认的配置，能够满足日常开发需求，如果需要修改，需要自己建立个配置文件

也就是说rabbitmq是不需要我们修改配置的。

一些常使用命令

• 查看rabbitmq状态   systemctl status rabbitmq

•  whereis rabbitmq 查看rabbitmq的目录

包括一些存放日志的地方

•  通过less rabbit@localhost.log 可以查询日志的目录等等

在bin中能看到下面的命令

 rabbitmqctl(8) — RabbitMQ

文档上面包括所有命令

RabbitMQ管理控制台

在rabbitmq中有管理插件，需要手动激活

rabbitmq-plugins enable rabbitmq_management

#添加用户 
rabbitmqctl add_user admin admin 
#为用户分配管理权限 
rabbitmqctl set_user_tags admin administrator 
#为用户分配资源权限 
https://www.rabbitmq.com/rabbitmqctl.8.html#set_permissions rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

• 创建和删除virtual hosts
• 查看、创建和删除users
• 查看创建和删除permissions
• 关闭其他用户的connections

访问web管理控制台

http://ip:15672

登录进去包括端口等等，连接 节点名称。 开的通道以及  通道的各种情况

虚拟主机

客户端使用

RabbitMQ Tutorials — RabbitMQ

提供给例子进行使用和操作

 客户端依赖

<dependency><groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId><version>5.7.3</version> 
</dependency>

使用场景用队列进行交互

利用队列进行交互

这里创建一个 生产者、以及消费者 直接利用ConnectionFactory ，并创建Channel 通道

• 创建连接工厂
• 设置连接属性
• 从连接工厂获取连接
• 从链接中创建通道
• 声明（创建）队列 如果队列不存在，才会创建 RabbitMQ 不允许声明两个队列名相同，属性不同的队列，否则会报错
• 发送消息

public class Producer {public static void main(String[] args) throws Exception {// 1、创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 2、设置连接属性factory.setHost("localhost");factory.setPort(5672);factory.setUsername("admin");factory.setPassword("admin");// factory.setVirtualHost(virtualHost);try (// 3、从连接工厂获取连接Connection connection = factory.newConnection("生产者");// 4、从链接中创建通道Channel channel = connection.createChannel();) {/*** 5、声明（创建）队列 如果队列不存在，才会创建 RabbitMQ 不允许声明两个队列名相同，属性不同的队列，否则会报错** queueDeclare参数说明：* * @param queue*            队列名称* @param durable*            队列是否持久化* @param exclusive*            是否排他，即是否为私有的，如果为true,会对当前队列加锁，其它通道不能访问，并且在连接关闭时会自动删除，不受持久化和自动删除的属性控制* @param autoDelete*            是否自动删除，当最后一个消费者断开连接之后是否自动删除* @param arguments*            队列参数，设置队列的有效期、消息最大长度、队列中所有消息的生命周期等等*/channel.queueDeclare("queue1", false, false, false, null);// 消息内容String message = "Hello World!";// 6、发送消息channel.basicPublish("", "queue1", null, message.getBytes());System.out.println("发送消息：" + message);}}
}

在消费者端

• 创建连接工厂
• 设置连接属性
• 从连接工厂获取连接
• 从链接中创建通道
• 声明（创建）队列 如果队列不存在，才会创建 RabbitMQ 不允许声明两个队列名相同，属性不同的队列，否则会报错
• 定义收到消息后的回调
• 开启队列消费 

public class Consumer {public static void main(String[] args) throws Exception {// 1、创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 2、设置连接属性factory.setHost("rabbitmq.study.com");factory.setUsername("admin");factory.setPassword("admin");String queueName = "queue1";try (// 3、从连接工厂获取连接Connection connection = factory.newConnection("消费者");// 4、从链接中创建通道Channel channel = connection.createChannel();) {/*** 5、声明（创建）队列 如果队列不存在，才会创建 RabbitMQ 不允许声明两个队列名相同，属性不同的队列，否则会报错** queueDeclare参数说明：* * @param queue*            队列名称* @param durable*            队列是否持久化* @param exclusive*            是否排他，即是否为私有的，如果为true,会对当前队列加锁，其它通道不能访问，*            并且在连接关闭时会自动删除，不受持久化和自动删除的属性控制。 一般在队列和交换器绑定时使用* @param autoDelete*            是否自动删除，当最后一个消费者断开连接之后是否自动删除* @param arguments*            队列参数，设置队列的有效期、消息最大长度、队列中所有消息的生命周期等等*/channel.queueDeclare(queueName, false, false, false, null);// 6、定义收到消息后的回调DeliverCallback callback = new DeliverCallback() {public void handle(String consumerTag, Delivery message) throws IOException {System.out.println("收到消息：" + new String(message.getBody(), "UTF-8"));}};// 7、开启队列消费channel.basicConsume(queueName, true, callback, new CancelCallback() {public void handle(String consumerTag) throws IOException {}});System.out.println("开始接收消息");System.in.read();}}
}

 在运行过后 web管理台上就可以看到非持久化的消息

数据和activemq还是不一样，默认时持久化的，没有消费者时，这个队列的消息是会一直存在的。

工作队列

消费者有多个的情况 Work Queue 当队列中的消息处理比较耗时时，我们可以开启多个消息者来并行消费处理消息。

只要没有确认，消息是一直存在的。

 这里在代码中使用是和上面的不一样的是 消息者

public static void main(String[] args) throws Exception {// 1、创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 2、设置连接属性factory.setHost("localhost");factory.setUsername("admin");factory.setPassword("admin");String queueName = "queue1";try (// 3、从连接工厂获取连接Connection connection = factory.newConnection("消费者1");// 4、从链接中创建通道Channel channel = connection.createChannel();// 可以同一连接创建多个通道，也可是不同连接创建通道 来组成多个消费者// Connection connection2 = factory.newConnection("消费者2");// Channel channel2 = connection2.createChannel();Channel channel2 = connection.createChannel();) {// 5、声明（创建）队列 如果队列不存在，才会创建 RabbitMQ 不允许声明两个队列名相同，属性不同的队列，否则会报错channel.queueDeclare(queueName, false, false, false, null);// 在消费者处理一个消息比较耗时时，减少预发来防止消息得不到及时处理channel.basicQos(1); // accept only one unack-ed message at a time// 6、定义收到消息后的回调DeliverCallback callback = (consumerTag, message) -> {System.out.println(consumerTag + " 收到消息：" + new String(message.getBody(), "UTF-8"));};// 7、开启队列消费channel.basicConsume(queueName, true, callback, consumerTag -> {});// 第二个消费者channel2.basicQos(1); // 只预取一个消息channel2.basicConsume(queueName, true, callback, consumerTag -> {});System.out.println("开始接收消息");}}

可以创建几个通道取消息，默认是256的预取消息，这里随便设置得值为1.每个通道取一次

在spring中使用

只需要在application.yml中添加下面得配置

spring:rabbitmq:host: rabbitmq.study.comport: 5672username: adminpassword: admin  # listener:#   simple:#      prefetch: 1

使用时添加对应得 RabbitListener  消费者

	@RabbitListener(queues = "hello")public void receive(String in) {System.out.println(" [x] Received '" + in + "'");}

生产者

34. 消息传递 (spring.io)   

• 这里配置一个我们要操作的Queue的bean,spring-rabbitmq框架在启动时将从容器中获取这些bean,
• 并向rabbitmq服务器创建这些queue、exchange、binding。
• 在RabbitAdmin.initialize()方法中做的这个事。
• 其完成的工作为：channel.queueDeclare("hello",false, false, false, null);
• 我们也可以自己手动通过 AmqpAdmin.declareXXX(xxx)方法来创建我们需要的queue、exchange、binding。

public class HelloWorldProducer {//spring boot 中 amqp的使用说明： //https://docs.spring.io/spring-boot/docs/2.1.6.RELEASE/reference/html/boot-features-messaging.html#boot-features-amqp/** 【注意】这里配置一个我们要操作的Queue的bean,spring-rabbitmq框架在启动时将从容器中获取这些bean,* 并向rabbitmq服务器创建这些queue、exchange、binding。* 在RabbitAdmin.initialize()方法中做的这个事。* 其完成的工作为：channel.queueDeclare("hello",false, false, false, null);* 我们也可以自己手动通过 AmqpAdmin.declareXXX(xxx)方法来创建我们需要的queue、exchange、binding。* * @Autowired private AmqpAdmin amqpAdmin;* * public void send() { * 		... * 		this.amqpAdmin.declareQueue(new Queue("hello"));* 		... * }*/@Beanpublic Queue hello() {return new Queue("hello");}// @Autowired  // private AmqpAdmin amqpAdmin;   //做queue、exchange、binding的管理用的@Autowiredprivate RabbitTemplate template;@Autowiredprivate Queue queue;@Scheduled(fixedDelay = 1000)   //定时多次发送消息public void send() {String message = "Hello World!";this.template.convertAndSend(queue.getName(), message);System.out.println(" [x] Sent '" + message + "'");}public static void main(String[] args) throws Exception {SpringApplication.run(HelloWorldProducer.class, args);System.in.read();}
}

也是为我们简化了写代码

然后在spring中设置 预取数，或者配置中添加都可以

	@Beanpublic SimpleRabbitListenerContainerFactory myFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();configurer.configure(factory, connectionFactory);// factory.setMessageConverter(myMessageConverter());factory.setPrefetchCount(1);return factory;}

publish/subscribe

Exchange 

• fanout：扇型交换机

channel.basicPublish("", "hello", null, message.getBytes());

channel.exchangeDeclare(String exchange, String type)

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));

public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setPort(5672);factory.setUsername("admin");factory.setPassword("admin");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel();Connection connection2 = factory.newConnection();Channel channel2 = connection2.createChannel();) {channel.exchangeDeclare(EXCHANGE_NAME, "fanout");// 创建一个临时队列，名字自动生成（唯一），连接断开，自动删除String queueName = channel.queueDeclare().getQueue();// 将队列绑定到exchange,绑定时指定的routingKey 也称// bingdingKey,在fanout交换器中routingKey无用。channel.queueBind(queueName, EXCHANGE_NAME, "");DeliverCallback deliverCallback = (consumerTag, message) -> {System.out.println(consumerTag + " 收到消息：" + new String(message.getBody(), "UTF-8"));};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});// 第二个消费者String queueName2 = channel2.queueDeclare().getQueue();channel2.queueBind(queueName2, EXCHANGE_NAME, "");channel2.basicConsume(queueName2, true, deliverCallback, consumerTag -> {});System.out.println("开始接收消息");System.in.read();}}

@Configuration
public class PubSubConfiguration {@Beanpublic FanoutExchange fanout() {return new FanoutExchange("spring-logs");}@Configurationpublic static class ReceiverConfig {@Beanpublic Queue autoDeleteQueue1() {return new AnonymousQueue();}@Beanpublic Queue autoDeleteQueue2() {return new AnonymousQueue();}@Beanpublic Binding binding1(FanoutExchange fanout, Queue autoDeleteQueue1) {return BindingBuilder.bind(autoDeleteQueue1).to(fanout);}@Beanpublic Binding binding2(FanoutExchange fanout, Queue autoDeleteQueue2) {return BindingBuilder.bind(autoDeleteQueue2).to(fanout);}}
}

channel.exchangeDeclare(EXCHANGE_NAME, "direct");String queueName = channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, "orange");DeliverCallback deliverCallback = (consumerTag, message) -> {System.out.println(consumerTag + " 收到消息：" + new String(message.getBody(), "UTF-8"));};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});// 第二个消费者String queueName2 = channel2.queueDeclare().getQueue();channel2.queueBind(queueName2, EXCHANGE_NAME, "black");channel2.queueBind(queueName2, EXCHANGE_NAME, "green");channel2.basicConsume(queueName2, true, deliverCallback, consumerTag -> {});

channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));

	@Beanpublic Binding binding1a(DirectExchange direct, Queue autoDeleteQueue1) {return BindingBuilder.bind(autoDeleteQueue1).to(direct).with("orange");}@Beanpublic Binding binding1b(DirectExchange direct, Queue autoDeleteQueue1) {return BindingBuilder.bind(autoDeleteQueue1).to(direct).with("black");}

channel.queueBind(queueName, "logs", "");

channel.queueBind(String queue, String exchange, String routingKey)

public static void main(String[] argv) {try (RPCClient fibonacciRpc = new RPCClient()) {for (int i = 0; i < 32; i++) {String i_str = Integer.toString(i);System.out.println(" [x] Requesting fib(" + i_str + ")");String response = fibonacciRpc.call(i_str);System.out.println(" [.] Got '" + response + "'");}} catch (IOException | TimeoutException | InterruptedException e) {e.printStackTrace();}}public String call(String message) throws IOException, InterruptedException {final String corrId = UUID.randomUUID().toString();String replyQueueName = channel.queueDeclare().getQueue();AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {if (delivery.getProperties().getCorrelationId().equals(corrId)) {response.offer(new String(delivery.getBody(), "UTF-8"));}}, consumerTag -> {});String result = response.take();channel.basicCancel(ctag);return result;}public void close() throws IOException {connection.close();}