SpringCloud之MQ笔记分享

MQ异步通信

初始MQ

同步通信

优点：时效性较强，可以以及得到结果

Feign就属于同步方式–问题：

• 耦合问题
• 性能下降（中间的等待时间）
• 资源浪费
• 级联失败

异步通信

优点

• 耦合度低
• 性能提升，吞吐量高
• 故障隔离
• 服务无强依赖，解决级联失败问题
• 流量削峰

缺点

• 依赖于broker的可靠性，安全性，吞吐能力
• 架构复杂了，业务没有明显的流程先，不好追踪管理

MQ常见框架

什么是MQ：信息队列，存放消息的队列，也就是时间驱动架构中的broker

	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司/社区	Rabbit	Apache	阿里	Apache
开发语言	Erlang	Java	Java	Scala&Java
协议支持	AMQP，XMPP，SMTP，STOMP	OpenWire,STOMP，REST,XMPP,AMQP	自定义协议	自定义协议
可用性	高	一般	高	高
单机吞吐量	一般	差	高	非常高
消息延迟	微秒级	毫秒级	毫秒级	毫秒以内
消息可靠性	高	一般	高	一般

推荐RabbitMQ

RabbitMQ快速入门

案例

发布者

package cn.itcast.mq.helloworld;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class PublisherTest {@Testpublic void testSendMessage() throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("xxx.xx.xxx.xxx");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("xxx");factory.setPassword("xxxxx");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.发送消息String message = "hello, rabbitmq1!";channel.basicPublish("", queueName, null, message.getBytes());System.out.println("发送消息成功：【" + message + "】");// 5.关闭通道和连接channel.close();connection.close();}
}

消费者

package cn.itcast.mq.helloworld;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ConsumerTest {public static void main(String[] args) throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("xxx.xxx.xx.xxx");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("xxx");factory.setPassword("xxx");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.订阅消息channel.basicConsume(queueName, true, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {// 5.处理消息String message = new String(body);System.out.println("接收到消息：【" + message + "】");}});System.out.println("等待接收消息。。。。");}
}

运行消费者后运行发布者，就可以接收到发布者的信息（因为消费者的函数是异步，所以发布者可以随时发，消费者一直等着）

SpringAMQP

什么是AMQP：一种协议，API规范

SpringAMQP：底层是rabbitMQ

基础实现

使用SpringAMQP实现Helllo World中的基础消息队列功能

1，引入依赖

<!--AMQP依赖，包含RabbitMQ-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2，基础配置

配置基础配置并且工具类并测试

基础配置

spring:rabbitmq:host: xxx.xx.xxx.xxport: 5672username: xxxpassword: xxxvirtual-host: /

测试类

package cn.itcast.mq.spring;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSendMessage(){String queueName = "simple.queue";String message = "hello,spring amqp";rabbitTemplate.convertAndSend(queueName,message);}
}

测试结果

前面几个是之前的，第三个和后面的是我自己设置的

上述为发布者，消费者需要：

1，配置yml文件

2，定义一个类

这里的核心是使用@Component自动装配到Spring,使用RabbitListener进行队列的监听，得到返回值msg就打印

package cn.itcast.mq.listener;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class SpringRabbitListener {@RabbitListener(queues = "simple.queue")public void listenSimpleQueue(String msg){System.out.println("消费者接收到消息：【"+msg+"】");}
}

WorkQueue模型

当一个消费者能力足够强，而另一个比较弱，合理的处理方式应该是强的消费者处理更多的消息，而弱的处理少的（根据能力分工）

所以WorkQueue模型就是为了让同一个队列的消费者，能力强的处理更多的消息，核心在于prefetch控制预提取的信息数量

生产者循环发送消息到simple.queue：发送消息

@Test
public void testWorkQueue() throws InterruptedException {
// 队列名称
String queueName = "simple.queue";
// 消息
String message = "hello, message__";
for (int i = 0; i < 50; i++) {
// 发送消息
rabbitTemplate.convertAndSend(queueName, message + i);// 避免发送太快Thread.sleep(20);}
}

消费者监听：两个消费者

@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException {
System.out.println("spring 消费者1接收到消息：【" + msg + "】");Thread.sleep(25);
}@RabbitListener(queues = "simple.queue") 
public void listenSimpleQueueMessage2(String msg) throws InterruptedException {
System.err.println("spring 消费者2接收到消息：【" + msg + "】");
Thread.sleep(100);
}

Work模型的使用：

• 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理

• 通过设置prefetch来控制消费者预取的消息数量

交换机：exchange

exchange：exchange负责消息路由，而不是存储，路由失败则消息丢失

FanoutExchange的使用

创建两个队列，并进行绑定

package cn.itcast.mq.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class FanoutConfig {//itcast.exchage@Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange("ylw.fanout");}//fanout.queue1@Beanpublic Queue fanoutQueue1() {return new Queue("fanout.queue1");}@Beanpublic Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}//fanout.queue2@Beanpublic Queue fanoutQueue2() {return new Queue("fanout.queue2");}@Beanpublic Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}

Consumer声明两个消费者

@RabbitListener(queues = "fanout.queue1")public void listenFanoutQueue1(String msg) 
{System.out.println("消费者1接收到Fanout消息：【" + msg + "】");}
@RabbitListener(queues = "fanout.queue2") public void listenFanoutQueue2(String msg) 
{System.out.println("消费者2接收到Fanout消息：【" + msg + "】");}

发布者发送消息

@Test
public void testFanoutExchange() {    // 队列名称    String exchangeName = "itcast.fanout";    // 消息String message = "hello, everyone!";    // 发送消息，参数分别是：交互机名称、RoutingKey（暂时为空）、消息     rabbitTemplate.convertAndSend(exchangeName, "", message);}