SpringCloud、Eureka、Zuul、Rabbitmq_Sun.niss

网络 02-07 3003

1.eureka注册中心

注册中心服务端主要对外提供了三个功能：服务注册：服务提供者启动时，会通过 Eureka Client 向 Eureka Server 注册信息，Eureka Server 会存储该服务的信息，Eureka Server 内部有二层缓存机制来维护整个注册表。提供注册表：服务消费者在调用服务时，如果 Eureka Client 没有缓存注册表的话，会从 Eureka Server 获取最新的注册表。同步状态： Eureka Client 通过注册、心跳机制和 Eureka Server 同步当前客户端的状态。

2.创建项目 2.1添加eureka依赖

2.2修改pom <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://·/jiahaobz/springcloud.git server: port: 6001 eureka: client: service-url: defaultZone: http://eureka1:2001/eureka,http://eureka2:2002/eureka

4.启动类加注解： @EnableConfigServer 5.访问测试： http://localhost:6001/item-service/dev http://localhost:6001/user-service/dev http://localhost:6001/order-service/dev 配置中心的客户端： 1.把2，3，4的application.yml注释掉

2.添加依赖： 03添加依赖：

<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-config</artifactId> </dependency>

3.添加新的配置文件：bootstrap.yml bootstrap.yml，引导配置，应用启动之前先执行 application.yml应用期待着之后执行

eureka: client: service-url: defaultZone: http://eureka1:2001/eureka,http://eureka2:2002/eureka spring: cloud: config: discovery: enabled: true service-id: config-service #下载配置文件 name: user-service profile: dev

检查确认： 1.按顺序启动项目： 05-eureka，等待完全启动 09-config，等待完全启动 03、04、06… 2.检查配置中心注册表中有config-server http://localhost:6001/user-service/dev http://localhost:6001/item-service/dev http://localhost:6001/order-service/dev 3.03-user的控制台日志，有没有连接6001服务器

10.RabbitMQ

消息队列、消息服务、消息中间件Broker 常见服务器：Rabbitmq、Activemq、Rocketmq、Kafka、Tubemq 搭建Rabbitmq服务器：设置vm网段：

ls ./ip-dhcp ifconfig

ip设置失败： nmcli n on systemcli restart NetworkManager

1.克隆一份虚拟机，docker-base，传文件 1.1从docker-base克隆一份虚拟机:rabbirmq 2.设置固定ip

./ip-static ip:192.168.64.140 ifconfig

3.上传文件到/root/

4.导入镜像

systemctl restart docker docker load -i rabbit-image.gz docker images

5.安装Rabbitmq： Docker 启动Rabbitmq 关闭防火墙:

systemctl stop firewalld systemctl disable firewalld # 重启 docker 系统服务 systemctl restart docker

配置管理员用户名和密码:

mkdir /etc/rabbitmq vim /etc/rabbitmq/rabbitmq.conf # 添加两行配置： default_user = admin default_pass = admin

启动Rabbitmq:

docker run -d --name rabbit \ -p 5672:5672 \ -p 15672:15672 \ -v /etc/rabbitmq/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \ -e RABBITMQ_CONFIG_FILE=/etc/rabbitmq/rabbitmq.conf \ --restart=always \ rabbitmq:management

访问管理控制台 http://192.168.64.140:15672 用户名密码是 admin

11.Rabbitmq创建

创建一个空的project，然后船舰一个maven工程：添加依赖：

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>cn.tedu</groupId> <artifactId>rabbitmq-api</artifactId> <version>1.0-SNAPSHOT</version> <dependencies> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.4.3</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.8.0</version> <configuration> <source>1.8</source> <target>1.8</target> </configuration> </plugin> </plugins> </build> </project>

创建服务者（发送消息）：

package m1; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Producer { public static void main(String[] args) throws IOException, TimeoutException { //连接服务器 ConnectionFactory f = new ConnectionFactory(); f.setHost("192.168.64.140"); f.setPort(5672); //5672用来收发消息，15672时管理控制台端口 f.setUsername("admin"); f.setPassword("admin"); Connection con = f.newConnection(); Channel c = con.createChannel(); //通信通道 //在服务器上创建一个队列:helloworld //如果队列在服务器上存在，不会重复创建 //第二个参数：是否是持久队列； //第三个参数：是否是排他队列、独占队列 //第四个参数：是否自动删除 //第五个参数：队列的其他属性 c.queueDeclare("helloworld",false,false,false,null); //向helloworld队列发送消息 c.basicPublish("", "helloworld", null, "Hello World".getBytes()); //不执行消息，关闭连接，关闭通信通道 c.close(); con.close(); } }

创建消费者（接收消息）：

package m1; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer { public static void main(String[] args) throws IOException, TimeoutException { //连接 ConnectionFactory f = new ConnectionFactory(); f.setHost("192.168.64.140"); f.setPort(5672); //5672用来收发消息，15672时管理控制台端口 f.setUsername("admin"); f.setPassword("admin"); Connection con = f.newConnection(); Channel c = con.createChannel(); //通信通道 //创建队列 c.queueDeclare("helloworld",false,false,false,null); System.out.println("等待接收数据"); //创建回调对象 DeliverCallback deliverCallback = (consumerTag, message) -> { byte[] a = message.getBody(); String s = new String(a); System.out.println("收到："+s); }; CancelCallback cancelCallback = consumerTag ->{}; //开始接收消息，把消息传递给一个回调对象进行处理 //第二个参数：是否自动确认(true)，autoAck c.basicConsume("helloworld", true,deliverCallback,cancelCallback); } }

合理分发： 1.让服务器可以知道消费者有没有处理完消息，手动确认 2. QPS=1，与抓取的消息数量，每次只接收一条消息，处理完之前不收下一条，手动确认模式才有效 3. 我们将任务封装为消息并将其发送到队列。后台运行的工作进程将获取任务并最终执行任务。当运行多个消费者时，任务将在它们之间分发。创建服务者：

package m2; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.Scanner; import java.util.concurrent.TimeoutException; public class Producer { public static void main(String[] args) throws IOException, TimeoutException { //连接服务器 ConnectionFactory f = new ConnectionFactory(); f.setHost("192.168.64.140"); f.setPort(5672); //5672用来收发消息，15672时管理控制台端口 f.setUsername("admin"); f.setPassword("admin"); Connection con = f.newConnection(); Channel c = con.createChannel(); //通信通道 //在服务器上创建一个队列:helloworld //如果队列在服务器上存在，不会重复创建 //第二个参数：是否是持久队列； //第三个参数：是否是排他队列、独占队列 //第四个参数：是否自动删除 //第五个参数：队列的其他属性 c.queueDeclare("helloworld", false, false, false, null); //向helloworld队列发送消息 while (true) { System.out.println("输入消息："); String s = new Scanner(System.in).nextLine(); c.basicPublish("", "helloworld", null, s.getBytes()); } //不执行消息，关闭连接，关闭通信通道 } }

创建消费者：

package m2; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer { public static void main(String[] args) throws IOException, TimeoutException { //连接 ConnectionFactory f = new ConnectionFactory(); f.setHost("192.168.64.140"); f.setPort(5672); //5672用来收发消息，15672时管理控制台端口 f.setUsername("admin"); f.setPassword("admin"); Connection con = f.newConnection(); Channel c = con.createChannel(); //通信通道 //创建队列 c.queueDeclare("helloworld",false,false,false,null); System.out.println("等待接收数据"); //创建回调对象 DeliverCallback deliverCallback = (consumerTag, message) -> { String s = new String(message.getBody()); System.out.println("收到："+s); //遍历访问每一个字符，每遇到一个‘.’，暂停一秒 for (int i = 0; i < s.length(); i++) { if (s.charAt(i)=='.'){ try { Thread.sleep(1000); }catch (Exception e){ e.printStackTrace(); } } } //发送回执 //1：回执，2：是否同时确认之前收到过的多条消息 c.basicAck(message.getEnvelope().getDeliveryTag(),false); System.out.println("...........消息处理完成"); }; CancelCallback cancelCallback = consumerTag ->{}; //QPS=1,每次受一条，处理完之前不收下一条，手动ack模式才有效 c.basicQos(1); //开始接收消息，把消息传递给一个回调对象进行处理 //第二个参数：是否自动确认(true)，autoAck c.basicConsume("helloworld", false,deliverCallback,cancelCallback); } }

消息持久化： 1.队列持久化

添加交换机：创建生产者：

package m3; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.Scanner; import java.util.concurrent.TimeoutException; public class Producer { public static void main(String[] args) throws IOException, TimeoutException { //连接服务器 ConnectionFactory f = new ConnectionFactory(); f.setHost("192.168.64.140"); f.setPort(5672); //5672用来收发消息，15672时管理控制台端口 f.setUsername("admin"); f.setPassword("admin"); Connection con = f.newConnection(); Channel c = con.createChannel(); //通信通道 //创建fanout交换机：logs //c.exchangeDeclare("logs", "fanout"); c.exchangeDeclare("logs", BuiltinExchangeType.FANOUT);//默认是非持久，FANOUT表示持久 //向交换机发送消息 while (true){ System.out.println("输入消息："); String s = new Scanner(System.in).nextLine(); //对交换机，第二个参数无效 c.basicPublish("logs", "", null, s.getBytes()); } } }

创建消费者：

package m3; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.UUID; import java.util.concurrent.TimeoutException; public class Consumer { public static void main(String[] args) throws IOException, TimeoutException { //连接 ConnectionFactory f = new ConnectionFactory(); f.setHost("192.168.64.140"); f.setPort(5672); //5672用来收发消息，15672时管理控制台端口 f.setUsername("admin"); f.setPassword("admin"); Connection con = f.newConnection(); Channel c = con.createChannel(); //通信通道 //1.创建队列 String queue = UUID.randomUUID().toString(); c.queueDeclare(queue,false,true,true,null); //2.创建交换机 c.exchangeDeclare("logs", BuiltinExchangeType.FANOUT); //3.绑定 c.queueBind(queue,"logs", ""); } } 12.BUS配置刷新，Rabbitmq整合到微服务中

在2、3、4、9添加BUS、Rabbitmq 1.在2、3、4、9添加依赖：rabbitmq、bus、binder-rabbit

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-bus</artifactId> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-rabbit</artifactId> </dependency>

09再单独加一个依赖：

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-actuator</artifactId> </dependency>

2.修改09的application.yml，添加Rabbitmq连接配置

spring: rabbitmq: host: 192.168.64.140 port: 5672 username: admin password: admin management: endpoints: web: exposure: include: bus-refresh

3.修改config目录中的三个配置文件，添加Rabbitmq连接配置在spring下一层添加

rabbitmq: host: 192.168.64.140 port: 5672 username: admin password: admin

4.提交推送到gitee仓库 5.启动：先等待05启动完成，等待09启动完成，然后访问测试：http://localhost:6001/order-service/dev，http://localhost:6001/item-service/dev，http://localhost:6001/user-service/dev 02 03 04 06启动后，查看控制台，要有连接6001服务器，再访问：http://localhost:6001/actuator ，查看有没有bus-refresh，有就用post请求：http://localhost:6001/actuator/bus-refresh

消息服务案例

1.bus配置刷新向rabbitmq发送刷新指令，其他模块接受指令并执行 2.sleuth+zipkin链路跟踪产生的链路跟踪日志发送到rabbitmq，zipkin从rabbitmq接收日志，简单模式 3.订单的流量削峰购物系统的订单，先发送到rabbitmq，后台消费者模块一个一个接收，存储订单，短时间内产生的大量订单，变成顺序处理，处理时间拉长。简单模式或工作模式

13. sleuth+zipkin链路跟踪 sleuth

用来产生链路监控日志，在2、3、4、6添加sleuth依赖，sleuth是0配置

<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-sleuth</artifactId> </dependency> zipkin

通过消息服务转发，解耦，流浪削峰， 2、3、4、6添加zipkin客户端，向rabbitmq发送日志： 1.在2、3、4、6添加zipkin客户端依赖

<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-zipkin</artifactId> </dependency>

2.在6中添加rabbitmq依赖

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>

3.在6中添加rabbitmq连接配置

rabbitmq: host: 192.168.64.140 port: 5672 username: admin password: admin

4.在2、3、4、6添加发送方式配置：rabbitmq

zipkin: sender: type: rabbit

下载zipkin服务器，开启服务器，连接rabbitmq： java -jar zipkin-server-2.23.16-exec.jar --zipkin.collector.rabbitmq.uri=amqp://admin:admin@192.168.64.140:5672 测试： http://localhost:9411/zipkin

14.eureka客户端选择正确网卡，注册ip地址

选择正确网卡，eureka客户端会自动选择网卡，可能会选择错误网卡进行注册，手动指定注册网卡：修改bootstrap.yml文件

spring: cloud: inetutils: ignored-interfaces: # 忽略的网卡 - VM.* #.任意字符 *0到多个 preferred-networks: # 要是用的网卡的网段 - 192\.168\.0\..+ # +是一到多个 15.拼多商城项目

2.springboot版本改成2.3.2RELEASE 3.如果数据库文件导入失败，执行下面sql语句，增大mysql缓存大小：

set global max_allowed_packet=100000000; set global net_buffer_length=100000; set global interactive_timeout=28800000; set global wait_timeout=28800000;

4.删除测试数据

delete from pd_user delete from pd_order delete from pd_order_item

5.选择SDK1.8 6.启动项目修改RunPdApp的启动配置，设置working directory工作目录：

购物系统生成的订单发送到rabbitmq

1.启动rabbitmq服务，可以重启docker容器 2.添加rabbitmq依赖

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> <version>2.6.2</version> </dependency>

3.yml添加rabbit连接配置

spring: rabbitmq: host: 192.168.64.140 port: 5672 username: admin password: admin #使用rabbitmq下自己空间 virtual-host: gjh

4.在启动类，或添加自动配置类里，添加队列参数配置

package com.pd.config; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * 配置使用的队列的参数 */ @Configuration public class QueueConfig { //RabbitAutoConfiguration自动配置类，会根据这里设置参数，在服务器上创建队列 @Bean public Queue orderQueue() { //持久，非独占，不自动删除 return new Queue("orderQueue", true, false, false); } }

5.在orderServiceImpl，注入AmqpTemplate工具，使用这个工具发送订单

复制一份pd-web改名为pd-web-consumer

修改端口为81，在consumer创建OrderConsumer类

package com.pd; import com.pd.pojo.PdOrder; import com.pd.service.OrderService; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; /** * 自动创建实例 * 自动注册成为消费 * 自动开始接收消息 * 自动处理收到的消息 * */ @Component //用来接收消息 @RabbitListener(queues = "orderQueue") public class OrderConsumer { @Autowired private OrderService orderService; @RabbitHandler //指定处理消息的方法,在同一个类中，只能设置一次 public void receive(PdOrder pdOrder) throws Exception { orderService.saveOrder(pdOrder); } } spring-boot整合rabbitmq

创建springboot项目添加依赖：修改配置文件，添加rabbitmq配置：

spring: rabbitmq: host: 192.168.64.140 port: 5672 username: admin password: admin #使用rabbitmq下自己空间 virtual-host: gjh 创建m1包

添加启动类：

@SpringBootApplication public class Main { public static void main(String[] args) { SpringApplication.run(Main.class, args); } //配置helloworld队列的参数 @Bean public Queue helloWorldQueue(){ return new Queue("helloworld",false,false,false); } @Autowired private Producer p; /* springboot 执行流程：扫描创建实例-->自动注入-->@PostConstruct-->后续流程 */ @PostConstruct public void test() { //在新的线程中执行阻塞 new Thread(() -> { try { Thread.sleep(3000); //等待消费者启动再发消息 } catch (InterruptedException e) { } p.send(); }).start(); } }

生产者：

@Component public class Producer { @Autowired private AmqpTemplate t; public void send(){ t.convertAndSend("helloworld","helloWorld!"); System.out.println("消息已发送"); } }

消费者：

/** * 自动创建实例 * 自动注册成为消费 * 自动开始接收消息 * 自动处理收到的消息 * */ @Component public class Consumer { @RabbitListener(queues = "helloworld") public void receive(String msg){ System.out.println("收到："+msg); } }

测试：

创建m2包（工作模式，轮询）

在主程序中创建名为**task_queue**的持久队列：

@SpringBootApplication public class Main { public static void main(String[] args) { SpringApplication.run(Main.class, args); } //配置htask_queue持久队列的参数 @Bean public Queue task_queue() { return new Queue("task_queue", true, false, false); } @Autowired private Producer p; /* springboot 执行流程：扫描创建实例-->自动注入-->@PostConstruct-->后续流程（启动消费者） */ @PostConstruct public void test() { //在新的线程中执行阻塞 new Thread(() -> { while (true){ System.out.println("输入消息："); String s = new Scanner(System.in).nextLine(); p.send(s); } }).start(); } }

生产者：

@Component public class Producer { @Autowired private AmqpTemplate t; public void send(String msg){ t.convertAndSend("task_queue",msg.getBytes()); System.out.println("消息已发送"); } }

消费者（创建两个）： @RabbitListener会创建一个消费者

@Component public class Consumer { @RabbitListener(queues = "task_queue") public void receive1(String msg){ System.out.println("消费者1收到："+msg); } @RabbitListener(queues = "task_queue") public void receive2(String msg){ System.out.println("消费者2收到："+msg); } } Ack模式：合理分发： 1.手动Ack Spring集成Rabbitmq，默认就是手动Ack，spring会自动发送回执 2.qos=1 yml配置添加prefetch参数，默认值250 消息持久化： 1.队列持久化 "task_queue", true, false, false 2.消息的持久化，spring默认已经添加持久参数

设置Ack模式，配置ynl文件：抓取数量：工作模式中, 为了合理地分发数据, 需要将 qos 设置成 1, 每次只接收一条消息, 处理完成后才接收下一条消息

spring: rabbitmq: listener: simple: prefetch: 1 创建m3包（发布和订阅模式）

广播消息，两个消费者同时能收到消息修改main类：创建队列修改创建交换机

@SpringBootApplication public class Main { public static void main(String[] args) { SpringApplication.run(Main.class, args); } //创建交换机 @Bean public FanoutExchange logs(){ //参数1：非持久参数2：不自动删除 return new FanoutExchange("logs",false,false); } @Autowired private Producer p; /* springboot 执行流程：扫描创建实例-->自动注入-->@PostConstruct-->后续流程（启动消费者） */ @PostConstruct public void test() { //在新的线程中执行阻塞 new Thread(() -> { while (true){ System.out.println("输入消息："); String s = new Scanner(System.in).nextLine(); p.send(s); } }).start(); } }

生产者：

@Component public class Producer { @Autowired private AmqpTemplate t; public void send(String msg){ t.convertAndSend("logs","",msg.getBytes()); System.out.println("消息已发送"); } }

消费者：

@Component public class Consumer { @RabbitListener(bindings = @QueueBinding( //随机队列，spring会自动随机命名，非持久、独占、自动删除(false,true,true) value = @Queue(), //交换机 //declare = "false"不在这里创建交换机，只用名称引用交换机 exchange = @Exchange(name = "logs",declare = "false") )) public void receive1(String msg){ System.out.println("消费者1收到："+msg); } @RabbitListener(bindings = @QueueBinding( //随机队列，spring会自动随机命名，非持久、独占、自动删除(false,true,true) value = @Queue(), //交换机 //declare = "false"不在这里创建交换机，只用名称引用交换机 exchange = @Exchange(name = "logs",declare = "false") )) public void receive2(String msg){ System.out.println("消费者2收到："+msg); } }

路由模式使用 direct 交换机队列和交换机绑定时, 设置**绑定键**发送消息时, 指定**路由键** main启动类：创建交换机为direct： @SpringBootApplication public class Main { public static void main(String[] args) { SpringApplication.run(Main.class, args); } //创建交换机 @Bean public DirectExchange logs(){ //参数1：非持久参数2：不自动删除 return new DirectExchange("direct_logs",false,false); } @Autowired private Producer p; /* springboot 执行流程：扫描创建实例-->自动注入-->@PostConstruct-->后续流程（启动消费者） */ @PostConstruct public void test() { //在新的线程中执行阻塞 new Thread(() -> { while (true){ System.out.println("输入消息："); String s = new Scanner(System.in).nextLine(); System.out.println("输入路由键："); String k = new Scanner(System.in).nextLine(); p.send(k,s); } }).start(); } }

生产者：

@Component public class Producer { @Autowired private AmqpTemplate t; public void send(String k,String msg){ t.convertAndSend("direct_logs",k,msg.getBytes()); System.out.println("消息已发送"); } }

消费者：

@Component public class Consumer { @RabbitListener(bindings = @QueueBinding( //随机队列，spring会自动随机命名，非持久、独占、自动删除(false,true,true) value = @Queue(), //交换机 //declare = "false"不在这里创建交换机，只用名称引用交换机 exchange = @Exchange(name = "direct_logs",declare = "false"), //绑定键关键词 key = {"error"} )) public void receive1(String msg){ System.out.println("消费者1收到："+msg); } @RabbitListener(bindings = @QueueBinding( //随机队列，spring会自动随机命名，非持久、独占、自动删除(false,true,true) value = @Queue(), //交换机 //declare = "false"不在这里创建交换机，只用名称引用交换机 exchange = @Exchange(name = "direct_logs",declare = "false"), //绑定键关键词 key = {"info","warning","error"} )) public void receive2(String msg){ System.out.println("消费者2收到："+msg); } }

测试：

主题模式

主题模式不过是具有特殊规则的路由模式, 代码与路由模式基本相同：

使用 topic 交换机使用特殊的绑定键和路由键规则 main启动类：

生产者：把交换机名字改为topic_logs 消费者：把交换机名字改为topic_logs 测试：