介绍Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上简而言之就是避免多个线程操作同个数据时发生数据不一致问题传统解决方法使用锁的方案可确保当一个线程在运行时,没有其他线程干扰,以下是常用的锁类型: 互斥锁 快速反馈锁 读写锁 Synchronized使用互斥锁,以确保同一时刻只有一个线程能够反复进入同步块java synchronized(this) { //执行逻辑... } 如果锁没有被其他线程占用，那么当前线程会获得锁并进入同步代码块或方法。 如果锁已经被其他线程占用，那么当前线程将进入等待状态，直到锁被释放。其他线程在释放锁后，等待队列中的线程将有机会竞争获取锁。 问题: 死锁：如果多个线程互相持有不同的锁，并且互相等待对方释放锁，就会发生死锁，导致所有线程都无法继续执行。 饥饿：某些线程可能会因为竞争较激烈，一直无法获得锁，从而导致饥饿现象，即某些线程无法得到执行的机会。 LockReentrantLocksynchronized的升级版,使用快速反馈锁,可对某个线程进行保护,直到该线程被解锁之前,其他线程无法进入,但被保护的线程可反复进入```java private Lock lock = new ReentrantLock();public void performTask() { lock.lock(); // 上锁 try { //执行逻辑 } finally { lock.unlock(); // 解锁 } } ```ReentrantLock区别与synchronized是当其他线程进了的时候,可以选择线程等待或是执行其他逻辑```java private Lock lock = new ReentrantLock();public void performTask() { if (lock.tryLock(3, TimeUnit.SECONDS)) {// 上锁,设置等待时间 try { //执行逻辑A } finally { lock.unlock(); // 解锁 } }else { //执行逻辑B } } ```ReentrantReadWriteLock和ReentrantReadWriteLock使用读写锁，它允许多个线程同时读取共享资源，但只允许一个线程写入共享资源。这在读多写少的情况下可以提高并发性能，因为多个线程可以同时读取资源，而写操作需要互斥```java private Lock lock = new ReentrantLock();public int readData() { readWriteLock.readLock().lock(); try { // 读取数据 return data; } finally { readWriteLock.readLock().unlock(); } }public void writeData(int newData) { readWriteLock.writeLock().lock(); // Acquire write lock try { // 写入数据 data = newData; } finally { readWriteLock.writeLock().unlock(); } } ```Redis分布式锁使用SETNX实现分布式锁powershell setnx key valye若key的值不存在,则执行set key value,若存在则不做任何操作```java String clientId = UUID.randomUUID().toString();Boolean success = redisTemplate.opsForValue().setIfAbsent("key", "locked",10,TimeUnit.SECONDS); try{ //执行逻辑 }finally{ if(clientId.equals(redisTemplate.opsForValue().get("key"))) redisTemplate.delete("key"); } ```Redssion分布式锁synchronized和Lock无法对分布式项目进行线程保护,因为传统方案使用的锁是JVM锁Redis分布式锁如果线程执行较长,key过期了线程还未结束,就出现操作同一数据问题Redssion对Redis分布式锁进行了封装,并使用定时任务,若key过期了线程还未结束,就自动加时间安装```xml org.springframework.boot spring-boot-starter-data-redis org.redisson redisson-spring-boot-starter 3.13.0 ```配置java @Bean public Redisson redisson(){ // 设置单机 Config config = new Config(); config.useSingleServer().setAddress("redis://localhost:6379"); return (Redisson) Redisson.create(config); }使用```java @Autowrite private Redisson redisson;public void performTask() { RLock redissonLock = redisson.getLock("key"); try { redissonLock.lock(); // 上锁 //执行逻辑 } finally { redissonLock.unlock();// 解锁 } } ```
初识MQ同步和异步通讯 同步通讯 微服务直接的直接调用就属于同步通讯优点:​ 时效性强,可立即得到结果缺点: 耦合性高​ 性能和吞吐能力低​ 有额外的资源消耗​ 有级联失效问题常用场景: 需要即可得到结果的服务之间使用同步通讯​ 异步通讯 异步通讯使用一个中间件Broker,当某个服务想要发起请求时,直接向Broker发起事件,然后继续执行自己的代码.Broker接收到事件后,会向其他服务发起调用,若该服务处于繁忙或宕机,则缓存等待后再发起.优点:​ 耦合度低​ 吞吐量提升​ 故障隔离​ 流量削峰缺点:​ 依赖于Broker的可靠性、安全性、吞吐能力​ 架构复杂,业务没有明显流程,不方便追钟问题MQ常见技术RabbitMQ概述和安装docker拉取镜像tex docker pull rabbitmq:3-management运行rabbitmq```text docker run \ -e RABBITMQ_DEFAULT_USER=用户名 \ -e RABBITMQ_DEFAULT_PASS=密码 \ -v mq-plugins:/plugins \ --name mq \ --hostname mq1 \ -p 15672:15672 \ -p 5672:5672 \ -d \ 镜像id```rabbit有两个端口,15672是管理员端口,从web可进入管理员页面,5672是用户端口,可从该端口使用rabbitmq消息模型rabbitMQ官方提供了6种消息模型,实际能用的有5种 基础队列模型 工作消息队列 广播 路由 主题 入门案例该入门案例使用的是官方的rabbitmq提供的连接方式,了解即可,后面使用spring提供的AMQP快捷连接导入依赖xml <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>3.0.4</version> </dependency>发送端```java // 1.建立连接 ConnectionFactory factory = new ConnectionFactory(); // 1.1设置连接参数,分别是:主机Ip、端口、vhost、用户名、密码 factory.setHost("主机IP"); factory.setPort(端口); factory.setVirtualHost("vhost"); factory.setUsername("用户名"); factory.setPassword("密码"); // 1.2建立连接 connection = factory.newConnection();// 2.创建通道 channel = connection.createChannel();// 3.创建队列 String queueName = "队列名"; channel.queueDeclare(queue, false, false, false, null);// 4.发送消息 String message = "hello,rabbitMQ"; channel.basicPublish("",queueName,null,message.getBytes());// 5.关闭通道和连接 channel.close(); connection.close(); ```接收端```java // 1.建立连接 ConnectionFactory factory = new ConnectionFactory(); // 1.1设置连接参数,分别是:主机Ip、端口、vhost、用户名、密码 factory.setHost("主机IP"); factory.setPort(端口); factory.setVirtualHost("vhost"); factory.setUsername("用户名"); factory.setPassword("密码"); // 1.2建立连接 connection = factory.newConnection();// 2.创建通道 channel = connection.createChannel();// 3.创建队列 String queueName = "队列名"; channel.queueDeclare(queue, false, false, false, null);// 4.订阅消息 channel.basicConsume(queueName,true,new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // 5.处理消息 String message = new String(body); System.out.println(message); } }); ```Spring AMQPspring AMQP是spring提供的一个可以快捷发送和接收mq消息的框架安装导入依赖xml <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>配置properties spring.rabbitmq.host=主机IP spring.rabbitmq.port=端口 spring.rabbitmq.virtual-host=vhost spring.rabbitmq.username=用户名 spring.rabbitmq.password=密码基础队列模型发送端```java @Autowired private RabbitTemplate rabbitTemplate;@GetMapping("/xx") rabbitTemplate.convertAndSend(队列名,消息); } ```接收端java @Component public class SpringRabbitListenner { @RabbitListener(queues = "test.queue") public void listenDemo(String message) throws InterruptedException{ System.out.println("收到rabbit消息:"+message); } }WorkQueue模型对比简单模型,如果服务1一次只能处理50条消息,而队列来了60条,那么就会有10条消息在队列等待那50条先处理完再处理,这样就会导致队列越来越多而服务器崩掉因此需要两个服务来接收队列消息,这样的操作就是WorkQueue工作模型WorkQueue是负载均衡发送,即只有一个消费者消费,如果想要发送给集群内所有服务,可以使用广播、路由java @RabbitListener(queues = "test.queue") public void listenDemo1(String message) throws InterruptedException{ System.out.println("收到rabbit消息1:"+message); } @RabbitListener(queues = "test.queue") public void listenDemo2(String message) throws InterruptedException{ System.out.println("收到rabbit消息2:"+message); }在工作模型中,服务会预取队列的消息,比如来了60条消息,就会平均分给两个服务器,而服务只会向队列里拿自己预取的消息一般情况下两个接收端是在不同的服务器下运行,若服务器1性能强一点,一次能处理多个,而服务器2性能弱一点,就不能使用预取```properties预取上限spring.rabbitmq.listener.simple.prefetch=1 ```Fanout(广播)模型fanout模型使用fanout交换机,发送端将消息推送(发送)至交换机里,由交换机分发至各个队列对比之前的简单模型和工作模型,它们一旦接收到到消息就会销毁消息,不会保存. 因此如果想一条消息发给多个服务,需要使用广播模型广播模型使用前需先创建队列和交换机,并把队列和交换机绑定创建和绑定交换机```java @Configuration public class FanoutConfig { // 创建Fanout交换机 @Bean public FanoutExchange fanoutExchange(){ return new FanoutExchange("fanouttest.exchange"); } // 创建队列1 @Bean public Queue fanoutQueue1(){ return new Queue("fanouttest.queue1"); } // 创建队列2 @Bean public Queue fanoutQueue2(){ return new Queue("fanouttest.queue2"); } //交换机和队列1绑定 @Bean public Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange); } //交换机和队列1绑定 @Bean public Binding fanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange); }} ```或者使用注解方式java @RabbitListener(bindings = @QueueBinding( value = @Queue(name = 队列名), exchange = @Exchange(name = 交换机名,type = ExchangeTypes.交换机类型), key = {key1,key2} )) public void listenDemo(String message) throws InterruptedException{ System.out.println(message); }推送(发送)和订阅(接收)​ 订阅和基础队列模型一样,只用推送有一点不同java rabbitTemplate.convertAndSend(交换机名,"",消息);Direct(路由)模型Direct Exchange会按照规则发送至指定的队列内在创建队列时,为队列设置好bindKey,交换机发送时,会发给对应的bindKey的队列一个队列可以有多个key,若两个队列key相同,则都发送,和广播模型效果相同,因此路由模型是广播模型的一种优化推送java rabbitTemplate.convertAndSend(交换机名,队列key,消息);订阅java @RabbitListener(bindings = @QueueBinding( value = @Queue(name = 队列名), exchange = @Exchange(name = 交换机名,type = ExchangeTypes.DIRECT), key = {key1,key2} )) public void listenDirecDemo2(String message) throws InterruptedException{ System.out.println("bule,yellow收到directest消息:"+message); }Topic(话题)模型Topic exchange也是按照一定规则发送至指定队列 相比Direct路由模型优势在于它的规则有点像正则表达式,可以用符号表达推送java rabbitTemplate.convertAndSend(交换机名,xx.yy,消息);订阅java @RabbitListener(bindings = @QueueBinding( value = @Queue(name = 队列名), exchange = @Exchange(name = 交换机名,type = ExchangeTypes.TOPIC), key = {#.yy,xx.#} )) public void Demo(String message) throws InterruptedException{ System.out.println(message); }消息转化AMQP发送和接收的类型都是Object,因此理论上不止String类型可发送,什么类型都可以发送 但是不推荐直接发送对象至队列里,因为如果要发送对象,必然要将对象序列化,而序列化后对象的字节数就变长了,导致传送过程变慢 因此推荐发送Json对象即可,Json对象本身就是序列化对象但是每次都要将对象转Json很麻烦,因此可以用一个转换器,在每次发送对象前,将对象转为Jsonjson依赖xml <dependency> <groupId>com.fasterxml.jackson.dataformat</groupId> <artifactId>jackson-dataformat-xml</artifactId> <version>2.13.3</version> </dependency>添加转换器在推送和订阅服务中添加转换器java @Bean public MessageConverter jsonMessageConverter(){ return new Jackson2JsonMessageConverter(); }消息可靠性确保消息能发送到消费者有以下情况会导致消息丢失 发送时丢失: 生产者发送的消息未到exchange 消息到达exchange后未到queue MQ宕机,queue将消息丢失 consumer收到消息后未消费就宕机 生产者消息确认RabbitMQ使用了一个publisher机制来避免消息丢失.当发送消息后,生产者会收到一个结果来确认发送状态 publisher-confirm,生产者确认 消息成功发送到交换机,返回ack 消息未发送到交换机,返回nack publisher-return,生产者回执 消息发送到交换机,但交换机没将消息发送到队列,返回ack,执行returncallback函数 首先，修改publisher服务中的application.yml文件，添加下面的内容：```yaml spring: rabbitmq: publisher-confirm-type: correlated publisher-returns: true template: mandatory: true```说明： publish-confirm-type：开启publisher-confirm，这里支持两种类型： simple：同步等待confirm结果，直到超时 correlated：异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback publish-returns：开启publish-return功能，同样是基于callback机制，不过是定义ReturnCallback template.mandatory：定义消息路由失败时的策略。true，则调用ReturnCallback；false：则直接丢弃消息 重写ReturnCallback交换机的消息到不了队列触发returncallback主要在RabbitTemplate里面,我们在启动spring时添加我们的配置,因此使用ApplicationContextAware,ApplicationContextAware接口会在bean准备好后执行setApplicationContextjava @Slf4j @Configuration public class CommonConfig implements ApplicationContextAware { @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { // 获取RabbitTemplate RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class); // 设置ReturnCallback rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> { // 投递失败，记录日志 log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息{}", replyCode, replyText, exchange, routingKey, message.toString()); // 如果有业务需要，可以重发消息 }); } }定义ConfirmCallback消息到不了交换机触发这个比较简单,在发送消息时可以定义对应的ConfirmCallback函数,这里使用内部类```java public void testSendMessage2SimpleQueue() throws InterruptedException { // 1.消息体 String message = "hello, spring amqp!"; // 2.全局唯一的消息ID，需要封装到CorrelationData中 CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString()); // 3.添加callback correlationData.getFuture().addCallback( result -> { if(result.isAck()){ // 3.1.ack，消息成功 log.debug("消息发送成功, ID:{}", correlationData.getId()); }else{ // 3.2.nack，消息失败 log.error("消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason()); } }, ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage()) ); // 4.发送消息 rabbitTemplate.convertAndSend("task.direct", "task", message, correlationData);// 休眠一会儿，等待ack回执 Thread.sleep(2000); } ```消息持久化未防止,MQ宕机,queue将消息丢失.我们可以将消息持久化到硬盘中默认情况下，SpringAMQP发出的任何交换机、队列、消息都是持久化的，不用特意指定。RabbitMQ安全保存主要分为三种: 交换机持久化 队列持久化 消息持久化 可以在RabbitMQ控制台看到持久化的都会带上D的标示：交换机持久化将交换机信息保存到硬盘,重启MQ后仍能找到交换机SpringAMQP中可以通过代码指定交换机持久化：java @Bean public DirectExchange simpleExchange(){ // 三个参数：交换机名称、是否持久化、当没有queue与其绑定时是否自动删除 return new DirectExchange("simple.direct", true, false); }队列持久化将队列信息保存到硬盘,重启MQ后仍能找到队列SpringAMQP中可以通过代码指定交换机持久化：java @Bean public Queue simpleQueue(){ // 使用QueueBuilder构建队列，durable就是持久化的 return QueueBuilder.durable("simple.queue").build(); }消息持久化将消息保存到硬盘,重启MQ后仍能找到消息```java public void testSendMessage2SimpleQueue() throws InterruptedException { // 1.消息体 String message = "hello, spring amqp!"; Message message = MessageBuilder .withBody("hello, spring amqp!".getBytes(StandardCharsets.UTF_8)) .setDeliveryMode(MessageDeliveryMode.PERSISTENT)//持久化 .build(); // 2.消息ID，需要封装到CorrelationData中 CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString()); // 3.发送消息 rabbitTemplate.convertAndSend("交换机", "队列", message, correlationData);// 休眠一会儿，等待ack回执 Thread.sleep(2000); } ```消费者消息确认RabbitMQ消息被消费者确认后会立刻删除,而SpringAMQP则允许配置三种确认模式,默认是none模式： manual：手动ack，需要在业务代码结束后，调用api发送ack,不推荐使用,容易被人恶意注入错误的api。 auto：自动ack，由spring监测listener代码是否出现异常，没有异常则返回ack；抛出异常则返回nack,而消息则不断尝试发送 none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除 none模式(默认)yaml spring: rabbitmq: listener: simple: acknowledge-mode: none # 关闭ackauto模式yaml spring: rabbitmq: listener: simple: acknowledge-mode: auto # 关闭ack出现异常listener会被不断执行,因为RabbitMQ消息未被确认,会一直尝试发送消费失败重试机制当消费者出现异常后，消息会不断requeue（重入队）到队列，再重新发送给消费者，然后再次异常，再次requeue，无限循环，导致mq的消息处理飙升，带来不必要的压力有两种方法解决失败重试机制: 本地重试:在本地重试,而不是MQ重试 失败策略:设置本地重试结束后执行的策略 本地重试开启本地重试时，消息处理过程中抛出异常，不会requeue到队列，而是在消费者本地重试,重试达到最大次数后，Spring会返回ack，消息会被丢弃yaml spring: rabbitmq: listener: simple: retry: enabled: true # 开启消费者失败重试 initial-interval: 1000 # 初识的失败等待时长为1秒 multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval max-attempts: 3 # 最大重试次数 stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false失败策略 RejectAndDontRequeueRecoverer:重试后向MQ发送确认,MQ会删除消息(默认) ImmediateRequeueMessageRecoverer:重试后向MQ发送nack，MQ会重新发送 RepublishMessageRecoverer:重试后向MQ发送确认,MQ会删除消息,同时向某交换机发送消息(推荐) 推荐使用RepublishMessageRecoverer,在业务上,当重试次数耗尽时,就代表这里出现了异常,可以将此异常通过交换机发送到对应的异常日志服务进行记录,然后人工介入时就可以快速找到问题java @Configuration public class ErrorMessageConfig { //定义重试结束后,发送的交换机 @Bean public DirectExchange errorMessageExchange(){ return new DirectExchange("error.direct"); } //定义重试结束后,发送的队列 @Bean public Queue errorQueue(){ return new Queue("error.queue", true); } //关联交换机和队列 @Bean public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){ return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error"); } //声明使用RepublishMessageRecoverer策略 @Bean public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){ return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error"); } }死信交换机(DLX)当交换机出现死信时,我们可以通过给交换机配置死信交换机把死信转发到死信交换机里产生死信的原因: 消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false(requeue禁止MQ重试) 消息是一个过期消息，超时无人消费 要投递的队列消息满了，无法投递 一般情况下,消息成为死信,MQ就会将消息丢弃如果配置了死信交换机,那么死信就不会丢弃,而是发送到死信交换机里与消费失败重试转异常交换机区别在于一个是从消费者发送到交换机,一个是从交换机发送到另一个交换机死信交换机功能: 和失败重试转异常交换机一样,出现异常时发送异常 消息过期时,发送过期消息到死信交换机 死信交换机配置:在消费者服务创建普通队列时,同时绑定交换机java // 声明普通的队列，并且为其指定死信交换机：dl.direct @Bean public Queue simpleQueue2(){ return QueueBuilder.durable("普通队列") // 指定队列名称 .deadLetterExchange("死信交换机") // 指定死信交换机 //.deadLetterRoutingKey("dl") // 指定key .build(); } // 声明死信交换机 dl.direct @Bean public DirectExchange dlExchange(){ return new DirectExchange("死信交换机", true, false); } // 声明存储死信的队列 dl.queue @Bean public Queue dlQueue(){ return new Queue("死信队列", true); } // 将死信队列 与 死信交换机绑定 @Bean public Binding dlBinding(){ return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple"); }过期时间TTLTTL就是一个消息的存活时间,当消息过期时就会被放入死信交换机TTL可以用来做延迟发送,就是将消息发给一个没有消费者的队列,超时后会转发给另一个有消费者的队列,从而实现延迟发送~生产者配置过期时间在发送消息时,指定消息的存活时间java @Test public void testTTLQueue() { // 创建消息 String message = "hello, ttl queue"; // 消息ID，需要封装到CorrelationData中 CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString()); // 发送消息 rabbitTemplate.convertAndSend("死信交换机", "死信队列", message, correlationData); // 记录日志 log.debug("发送消息成功"); }消费者配置过期时间在创建队列时,设置消息的存活时当生产者与消费者同时配置TTL,以最短的为准java // 声明普通的 simple.queue队列，并且为其指定死信交换机：dl.direct @Bean public Queue simpleQueue2(){ return QueueBuilder.durable("普通队列") // 指定队列名称，并持久化 .ttl(10000) // 设置队列的超时时间，10秒 .deadLetterExchange("死信交换机") // 指定死信交换机 //.deadLetterRoutingKey("dl") // 指定key .build(); } // 声明死信交换机 dl.direct ... // 声明存储死信的队列 dl.queue ... // 将死信队列 与 死信交换机绑定 ... }延迟队列我们在业务中经常遇到以下场景,这时需要用上延迟队列: 延迟发送短信 用户下单，如果用户在15 分钟内未支付，则自动取消 预约工作会议，20分钟后自动通知所有参会人员 我们想要实现消息的延迟发送,可以使用以下方案: TTL,将消息发送给没有消费者的队列,超时后转发 延迟插件 DelayExchange插件安装可以去对应的GitHub页面下载3.8.9版本的插件，地址为https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/tag/3.8.9这个对应RabbitMQ的3.8.5以上版本。基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷。之前设定的RabbitMQ的数据卷名称为mq-plugins，所以我们使用下面命令查看数据卷：sh docker volume inspect mq-plugins可以得到下面结果：接下来，将插件上传到这个目录即可：最后就是安装了，需要进入MQ容器内部来执行安装。我的容器名为mq，所以执行下面命令：sh docker exec -it mq bash执行时，请将其中的 -it 后面的mq替换为你自己的容器名.进入容器内部后，执行下面命令开启插件：sh rabbitmq-plugins enable rabbitmq_delayed_message_exchange结果如下：网站使用在网站内直接创建交换机,类型未x-deleyed-message在发消息的时候需要指定消息头x-delay和时间springAMQP使用基于注解方式（推荐）：java @RabbitListener(bindings = @QueueBinding( value = @Queue(name = 队列名), exchange = @Exchange(name = "交换机类型.名称",delayed=ture), key = {#.yy,xx.#} )) public void Demo(String message) throws InterruptedException{ System.out.println(message); }也可以基于@Bean的方式：java @Configuration public class DirectConfig { // 创建交换机 @Bean public DirectExchange directExchange(){ return new ExchangeBuilder .directExchange("交换机类型.名称") .delayed()//开启插件 .durable(ture)//持久化 .builder; } // 创建队列 @Bean ... //交换机和队列绑定 @Bean ... } }消息生产者发送消息时指定消息头x-delay和时间java public void testSendMessage2SimpleQueue() throws InterruptedException { // 1.消息体 String message = "hello, spring amqp!"; Message message = MessageBuilder .withBody("hello, spring amqp!".getBytes(StandardCharsets.UTF_8)) .setDeliveryMode(MessageDeliveryMode.PERSISTENT)//持久化 .setHeader("x-delay",int毫秒) .build(); // 2.消息ID，需要封装到CorrelationData中 CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString()); // 3.发送消息 rabbitTemplate.convertAndSend("交换机", "队列", message, correlationData); }惰性队列当生产者发送消息的速度超过了消费者处理消息的速度，就会导致队列中的消息堆积，直到队列存储消息达到上限。之后发送的消息就会成为死信，可能会被丢弃，这就是消息堆积问题 增加更多消费者，提高消费速度。也就是我们之前说的work queue模式 增加消费者性能,比如使用多线程 扩大队列容积，提高堆积上限,在安装MQ时配置,默认一个队列大小上限是内存的40% 使用惰性队列,将消息存到磁盘而非硬盘 修改队列为惰性设置一个队列为惰性队列，只需要在声明队列时，指定x-queue-mode属性为lazy即可。可以通过命令行将一个运行中的队列修改为惰性队列：sh rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues命令解读： rabbitmqctl ：RabbitMQ的命令行工具 set_policy ：添加一个策略 Lazy ：策略名称，可以自定义 "^lazy-queue$" ：用正则表达式匹配队列的名字 '{"queue-mode":"lazy"}' ：设置队列模式为lazy模式 --apply-to queues：策略的作用对象，是所有的队列 创建惰性队列需要注意,队列和交换机(有用上交换机的话),必须持久化才能使用惰性队列基于注解方式（推荐）：java @RabbitListener(bindings = @QueueBinding( value = @Queue(name = 队列名), durable = "ture", arguments = @Argument(name = "x-queue-mode",model = "lazy") )) public void Demo(String message) throws InterruptedException{ System.out.println(message); }也可以基于@Bean的方式：java @Configuration public class DirectConfig { // 创建交换机 @Bean ... // 创建队列 @Bean public LazyQueue lazyQueue(){ return new QueueBuilder .durable("队列名") .lazy()//开启惰性队列 .builder;//持久化 } //交换机和队列绑定 @Bean ... } }MQ集群RabbitMQ的是基于Erlang语言编写，而Erlang又是一个面向并发的语言，天然支持集群模式。RabbitMQ的集群有两种模式： 普通集群：是一种分布式集群，将队列分散到集群的各个节点，从而提高整个集群的并发能力。 镜像集群：是一种主从集群，普通集群的基础上，添加了主从备份功能，提高集群的数据可用性。 镜像集群虽然支持主从，但主从同步并不是强一致的，某些情况下可能有数据丢失的风险。因此在RabbitMQ的3.8版本以后，推出了新的功能：仲裁队列来代替镜像集群，底层采用Raft协议确保主从的数据一致性。普通集群普通集群，或者叫标准集群（classic cluster），具备下列特征： 会在集群的各个节点间共享部分数据，包括：交换机、队列元信息。不包含队列中的消息。 当访问集群某节点时，如果队列不在该节点，会从数据所在节点传递到当前节点并返回 队列所在节点宕机，队列中的消息就会丢失 结构如图：我们先在之前启动的mq容器中获取一个cookie值，作为集群的cookie。执行下面的命令：sh docker exec -it mq cat /var/lib/rabbitmq/.erlang.cookie可以看到cookie值如下：sh FXZMCVGLBIXZCDEMMVZQ接下来，停止并删除当前的mq容器，我们重新搭建集群。sh docker rm -f mq在/tmp目录新建一个配置文件 rabbitmq.conf：```sh cd /tmp创建文件touch rabbitmq.conf ```文件内容如下：nginx loopback_users.guest = false listeners.tcp.default = 5672 cluster_formation.peer_discovery_backend = rabbit_peer_discovery_classic_config cluster_formation.classic_config.nodes.1 = rabbit@mq1 cluster_formation.classic_config.nodes.2 = rabbit@mq2 cluster_formation.classic_config.nodes.3 = rabbit@mq3再创建一个文件，记录cookie```sh cd /tmp创建cookie文件touch .erlang.cookie写入cookieecho "FXZMCVGLBIXZCDEMMVZQ" > .erlang.cookie修改cookie文件的权限chmod 600 .erlang.cookie ```准备三个目录,mq1、mq2、mq3：```sh cd /tmp创建目录mkdir mq1 mq2 mq3 ```然后拷贝rabbitmq.conf、cookie文件到mq1、mq2、mq3：```sh进入/tmpcd /tmp拷贝cp rabbitmq.conf mq1 cp rabbitmq.conf mq2 cp rabbitmq.conf mq3 cp .erlang.cookie mq1 cp .erlang.cookie mq2 cp .erlang.cookie mq3 ```创建一个网络：sh docker network create mq-net运行命令sh docker run -d --net mq-net \ -v ${PWD}/mq1/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \ -v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \ -e RABBITMQ_DEFAULT_USER=itcast \ -e RABBITMQ_DEFAULT_PASS=123321 \ --name mq1 \ --hostname mq1 \ -p 8071:5672 \ -p 8081:15672 \ rabbitmq:3.8-managementsh docker run -d --net mq-net \ -v ${PWD}/mq2/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \ -v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \ -e RABBITMQ_DEFAULT_USER=itcast \ -e RABBITMQ_DEFAULT_PASS=123321 \ --name mq2 \ --hostname mq2 \ -p 8072:5672 \ -p 8082:15672 \ rabbitmq:3.8-managementsh docker run -d --net mq-net \ -v ${PWD}/mq3/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \ -v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \ -e RABBITMQ_DEFAULT_USER=itcast \ -e RABBITMQ_DEFAULT_PASS=123321 \ --name mq3 \ --hostname mq3 \ -p 8073:5672 \ -p 8083:15672 \ rabbitmq:3.8-management如图，在mq2和mq3两个控制台也都能看到：镜像集群镜像集群：本质是主从模式，具备下面的特征： 交换机、队列、队列中的消息会在各个mq的镜像节点之间同步备份。 创建队列的节点被称为该队列的主节点，备份到的其它节点叫做该队列的镜像节点。 一个队列的主节点可能是另一个队列的镜像节点 所有操作都是主节点完成，然后同步给镜像节点 主宕机后，镜像节点会替代成新的主 结构如图：和普通集群部署一样,在部署好后通过docker进入任意一个节点进行配置即可:sh docker exec -it mq1 bash镜像模式的配置有3种模式：| ha-mode | ha-params | 效果 | | :-------------- | :---------------- | :----------------------------------------------------------- | | 准确模式exactly | 队列的副本量count | 集群中队列副本（主服务器和镜像服务器之和）的数量。count如果为1意味着单个副本：即队列主节点。count值为2表示2个副本：1个队列主和1个队列镜像。换句话说：count = 镜像数量 + 1。如果群集中的节点数少于count，则该队列将镜像到所有节点。如果有集群总数大于count+1，并且包含镜像的节点出现故障，则将在另一个节点上创建一个新的镜像。 | | all | (none) | 队列在群集中的所有节点之间进行镜像。队列将镜像到任何新加入的节点。镜像到所有节点将对所有群集节点施加额外的压力，包括网络I / O，磁盘I / O和磁盘空间使用情况。推荐使用exactly，设置副本数为（N / 2 +1）。 | | nodes | node names | 指定队列创建到哪些节点，如果指定的节点全部不存在，则会出现异常。如果指定的节点在集群中存在，但是暂时不可用，会创建节点到当前客户端连接到的节点。 |这里我们以rabbitmqctl命令作为案例来讲解配置语法。语法示例：exactly模式rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}' rabbitmqctl set_policy：固定写法 ha-two：策略名称，自定义 "^two\."：匹配队列的正则表达式，符合命名规则的队列才生效，这里是任何以two.开头的队列名称 '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}': 策略内容 "ha-mode":"exactly"：策略模式，此处是exactly模式，指定副本数量 "ha-params":2：策略参数，这里是2，就是副本数量为2，1主1镜像 "ha-sync-mode":"automatic"：同步策略，默认是manual，即新加入的镜像节点不会同步旧的消息。如果设置为automatic，则新加入的镜像节点会把主节点中所有消息都同步，会带来额外的网络开销 all模式rabbitmqctl set_policy ha-all "^all\." '{"ha-mode":"all"}' ha-all：策略名称，自定义 "^all\."：匹配所有以all.开头的队列名 '{"ha-mode":"all"}'：策略内容 "ha-mode":"all"：策略模式，此处是all模式，即所有节点都会称为镜像节点 nodes模式rabbitmqctl set_policy ha-nodes "^nodes\." '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}' rabbitmqctl set_policy：固定写法 ha-nodes：策略名称，自定义 "^nodes\."：匹配队列的正则表达式，符合命名规则的队列才生效，这里是任何以nodes.开头的队列名称 '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}': 策略内容 "ha-mode":"nodes"：策略模式，此处是nodes模式 "ha-params":["rabbit@mq1", "rabbit@mq2"]：策略参数，这里指定副本所在节点名称 仲裁队列仲裁队列是镜像模式的替代方案,使用数据强一致方案,有效防止了数据丢失问题网站使用SpringAMQP使用java @Bean public Queue quorumQueue() { return QueueBuilder .durable("quorum.queue") // 持久化 .quorum() // 仲裁队列 .build(); }SpringAMQP使用集群注意，这里用address来代替host、port方式java spring: rabbitmq: addresses: 192.168.150.105:8071, 192.168.150.105:8072, 192.168.150.105:8073 username: itcast password: 123321 virtual-host: /
微服务介绍单体架构单体式架构将所有服务放在同一个项目 部署简单 耦合性高 分布式架构分布式价格将多种服务拆分到不同项目 部署麻烦 耦合性低 高可用 集群部署困难 微服务的分布式架构微服务的分布式在原有的分布式架构上进行改良,解决了传统分布式中部署和集群等问题常见的分布式技术Spring Cloud项目结构在微服务项目中,不同的功能应放在不同的项目里远程调用不同的服务之间数据需要通信时,由于各服务已经暴露了接口位置,所以当A服务想要使用B服务的功能时,可使用http通信方式调用HttpUtil这是比较传统Http通信,需要自行创建HttpUtil,再调用```java public class HttpUtil { private static String USER_AGENT = "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 6.0; Trident/4.0; SLCC1; .NET CLR 2.0.50727; .NET CLR 3.0.04506; customie8)";// HTTP GET request public static String sendGet(String url, String charset) throws Exception { URL realurl = new URL(url); HttpURLConnection con = (HttpURLConnection) realurl.openConnection(); // optional default is GET con.setRequestMethod("GET"); // add request header con.setRequestProperty("User-Agent", USER_AGENT); // int responseCode = con.getResponseCode(); // System.out.println("Response Code : " + responseCode); BufferedReader in = new BufferedReader(new InputStreamReader(con.getInputStream(), charset)); String inputLine; StringBuffer result = new StringBuffer(); while ((inputLine = in.readLine()) != null) { result.append(inputLine); result.append("\r\n"); } in.close(); con.disconnect(); return result.toString(); } // HTTP POST request @SuppressWarnings("deprecation") public static String sendPost(String url, Map<String, String> param, String charset) throws Exception { URL realurl = new URL(url); HttpsURLConnection con = (HttpsURLConnection) realurl.openConnection(); con.setRequestMethod("POST"); // add reuqest header con.setRequestProperty("User-Agent", USER_AGENT); // con.setRequestProperty("Accept-Language", "en-US,en;q=0.5"); // String urlParameters = // "sn=C02G8416DRJM&cn=&locale=&caller=&num=12345"; StringBuffer buffer = new StringBuffer(); if (param != null && !param.isEmpty()) { for (Map.Entry<String, String> entry : param.entrySet()) { buffer.append(entry.getKey()).append("=").append(URLEncoder.encode(entry.getValue(), charset)) .append("&"); } } buffer.deleteCharAt(buffer.length() - 1); // Send post request con.setDoOutput(true); DataOutputStream wr = new DataOutputStream(con.getOutputStream()); wr.writeBytes(buffer.toString()); wr.flush(); wr.close(); // int responseCode = con.getResponseCode(); // System.out.println("Post parameters : " + urlParameters); // System.out.println("Response Code : " + responseCode); BufferedReader in = new BufferedReader(new InputStreamReader(con.getInputStream(), charset)); String inputLine; StringBuffer result = new StringBuffer(); while ((inputLine = in.readLine()) != null) { result.append(inputLine); result.append("\r\n"); } in.close(); con.disconnect(); return result.toString(); } } ```java String json = HttpUtil.sendGet(accessTokenUrl, "utf-8");RestTemplateRestTemplate是一种基于spring的http通信方式使用前需要注册java @Bean public RestTemplate restTemplate(){ return new RestTemplate(); }java User user = restTemplate.getForObject(url,User.class)FeginFegin比RestTemplate更加优雅,且兼容了负载均衡,无需更多配置安装 导入依赖 xml <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-openfeign</artifactId> </dependency> 添加注解@EnableFeignClients在application中 ```java @SpringBootApplication @EnableFeignClients public class UserServiceApplication { public static void main(String[] args) { SpringApplication.run(UserServiceApplication.class, args); } } ```使用java @FeginClient("ip:端口或nacose服务名") public interface xxClint{ @请求方式Mapping("请求路径") 返回值类 xx(请求参数类 请求参数) }自定义配置Fegin可以通过运行自定义配置来覆盖默认配置有两种方法可以修改配置 基于配置文件 properties fegin.client.config.default.类型 = 值 基于Bean和注解 先写好对应的Beanjava @Configuration public class FeginClientConfiguration{ @Bean public 类型(){ return 类型.值; } } 如果是全局配置,把它放到@EnableFeignClientsjava @EnableFeignClients(defaultConfiguration = FeginClientConfiguration.class) 如果是局部配置,把它放到@FeginClientjava @FeginClient(value="ip:端口" ,defaultConfiguration = FeginClientConfiguration.class) 优化 日志尽量使用NONE 使用连接池 导入依赖xml <dependency> <groupId>io.github.openfeign</groupId> <artifactId>feign-httpclient</artifactId> </dependency>配置properties #开启配置 feign.httpclient.enabled=true #最大连接数 feign.httpclient.max-connections=200 #每个路径最大连接数 feign.httpclient.max-connections-per-route=50最佳实践在实际开发中,同个前人不断测试,有几种方案开发方式更受开发人员接收,这里介绍抽取Feign方法,用这个方法,其他的服务共用一个client和pojo 创建一个model,命名为feign-api,引入依赖(或在父依赖里引入) 将clients和pojo放入该model里写 其他服务若要使用,引入feign-api依赖 由于不像dubbo一样是有公共spring容器,因此还需要导入对应的api进容器里 方法一:指定Feign所在的包java @EnableFeignClients(basePackages = 依赖所在的包)方法二:指定Feign字节码java @EnableFeignClients(clients = {xxClient.class})Eureka注册中心如果某个服务是采用集群的方式,它就有多个端口,那么消费者如何才能找到正确的端口,这时就需要注册中心动态管理Eureka原理Eureka搭建搭建注册中心创建一个新的项目,和其他服务同级,并导入mavenxml <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-eureka-server</artifactId> </dependency>在eureka服务的application中配置java @SpringBootApplication @EnableEurekaServer public class EurakaAppliacation { public static void main(String[] args) { SpringApplication.run(EurakaAppliacation.class,args); } }```properties server.port=10086 spring.application.name=eurakaserviceeureka地址信息eureka.client.service-url.defaultZone:http://localhost:10086/eureka ```服务注册服务注册仅需在配置文件中写明euraka地址```properties应用名称spring.application.name=user-service server.port=8081eureka地址信息eureka.client.service-url.defaultZone:http://localhost:10086/eureka ```当一个服务需要部署集群,多次注册时,可以使用idea中的copy服务,但是一定要写明端口不同tex -Dservice.port=端口服务发现使用远程调用(这里以RestTemplate为例)java @Bean @LoadBalanced public RestTemplate restTemplate(){ return new RestTemplate(); }java String url="http://orderservice/order/getOrder"; String str = restTemplate.getForObject(url,String.class);注意 当使用微服务接口时,RestTemplate不能使用localhost或者ip调用服务,只能使用服务名称调用 服务名称有下划线_时,url将下划线_改为减号- RestTemplate访问微服务接口,在注册时要添加@LoadBalanced注解表示负载均衡 整体流程(简略)Ribbon负载均衡远程调用服务地址是无法通过http直接访问的,通过使用@LoadBalanced注解,可生成一个拦截器,在发送http请求时,会拉去对应的服务曲找eureka寻找对应的ip和端口,在使用Ribbon进行负载均衡负载均衡策略负载均衡是通过一个叫IRule的接口来实现的,每个子接口都代表一个规则| 负载均衡规则类 | 规则描述 | | :------------: | ------------- | | RoundRobinRule | 轮询,默认策略 | | RandomRule | 随机 | | RetryRule | 轮询重试 | | WeightedResponseTimeRule | 响应速度决定权重,权重越大则选中的概率越大||BestAvailableRule|使用并发量最小| |AvailabilityFilteringRule|先过滤不可用的,再使用并发量最小| |ZoneAvoidanceRule|以区域可用的服务器为基础进行服务器选择.使用Zone对服务器进行分类.Zone可理解为一个机房.而后在对Zone进行轮询|负载均衡定义 代码方式 java @Bean public Irule 负载均衡规则类(){ return new 负载均衡规则类(); }服务使用此方法定义负载均衡策略,在该服务下调用的所有服务都是基于此策略进行负载均衡 配置文件 properties 被调用的服务名.ribbon.NFLoadBalancerRuleClassName = com.netflix.loadbalancer.负载均衡规则类使用此方法,服务之间是一对一的负载均衡策略饥饿加载Ribbon默认采用的是懒加载,即第一次访问才会去创建DynamicServiceListLoadBalancer,请求时间会很长,而饥饿加载会在服务启动时创建,从而降低第一期请求时间properties ribbon.eureka.enabled=true ribbon.eager-load.clients=xxservice,xxservice如果第一次访问还是慢,可能是springboot的DispathService也是懒加载,这些也要配Nacos注册中心Nacos是阿里巴巴开发的一个注册中心管理模块,现在是springCloud的一个组件,它的功能比Eureka更丰富,在国内更受欢迎搭建注册中心在nacos.io网站上去访问github,然后下载最新版,有win和liunx版本,根据自己选择下载打开nacos后,可在http://192.168.18.1:8848/nacos/index.html中查看在pom文件的dependencyManagement里,也就是spring-cloud-dependencies下面添加spring-cloud-alibaba依赖管理xml <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-alibaba-dependencies</artifactId> <version>2.2.5.RELEASE</version> <type>pom</type> <scope>import</scope> </dependency>然后添加依赖nacos依赖,若有其他注册中心依赖,需要把其他注册中心注释掉xml <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId> </dependency>服务注册或发现```properties应用名称spring.application.name=user-service server.port=8081nacos地址信息,默认localhost:8848spring.cloud.nacos.server-addr=localhost:8848 ```多级存储模型(集群)在实际应用中,不同的服务是在不同的地区服务器执行的,这个时候可以对不同的服务进行集群管理在配置文件中添加集群名称即可在nacos管理中心查看```properties配置集群spring.cloud.nacos.discovery.cluster-name= FZ ```负载均衡策略(eureka基础上附加)| 负载均衡规则类 | 规则描述 | | :------------: | -------------------- | | NacosRule | 优先随机访问相同集群 |负载均衡定义和Eureka负载均衡定义相同,不过如果使用NacosRule,它的位置和其他负载均衡规则类不同properties 被调用的服务名.ribbon.NFLoadBalancerRuleClassName = com.alibaba.cloud.nacos.ribbon.NacosRule负载均衡权重NacosRule虽然是随机策略,但也有权重,可以在nacos管理中心设置权重设置权重几点好处: 性能差的服务器权重低,并发量也就会低 若服务器需要更新,可以先将部分服务权重调0,更新好后再上调一点,放客户进来测试,最后调整正常,更新其他服务 环境隔离在实际开发中,我们需要将项目分为开发、测试和上线环境,不同环境的服务是不能互相访问的,这就需要使用环境隔离首先需要先创建Namespce命名空间然后在配置文件中设置对应的命名空间idproperties spring.cloud.nacos.discovery.namespace=命名空间id注册中心Nacos和Eureka的区别 Nacos会将服务分为零时实例和非零时实例 零时示例和Eureka的服务一样,采用心跳检测,每隔30s查看服务提供者是否宕机.非零时示例只用当需要想服务提供者请求服务时,才询问是否宕机,若宕机则等待,和zookeeper方式一样零时和非临时示例配置properties spring.cloud.nacos.discovery.ephemeral=true Nacos和Eureka的服务一样,定时获取到对应的服务ip后,会将ip缓存,但是Nacos多了一个功能就是当心跳检测发现其宕机后,会主动告诉消费者,让ip缓存清除 Nacos配置管理服务在实际开发中,若项目由多个服务,那么对各服务的配置一个个管理就显得麻烦,且配置完成后要重启,带来不必要损失,这时候我们就需要热更新配置Nacos提供了一种配置管理服务,各服务读取配置管理服务,获取对应的配置但是在实际开发中,为了节约成本,我们会把注册中心服务和配置管理服务放在一个服务里创建配置在Nacos管理页中添加配置在设置配置ID时,有一个约定成俗的命名,服务名称-开发环境.xxx获取配置信息引入Nacos配置管理依赖properties <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId> </dependency>然后再bootstrap.properties/yaml中写固定配置```properties应用名称spring.application.name=服务名称开发环境spring.profiles.active=开发环境配置文件类型spring.cloud.nacos.config.file-extension=服务后最nacos ip和地址spring.cloud.nacos.server-addr=nacosIp:端口应用服务 WEB 访问端口server.port=服务端口 ```这样,以后的所有配置都可以在nacos配置管理中心写,不用在application.properties/yaml中写了但是实际开发中,我们一般把可能会需要改变的写在nacos中,固定不变的可以写在Bootstrap或application里配置热更新配置热更新方式有两种 基于注解@RefreshScope ​ 在controller中添加,并使用@value注入java @RestController @RefreshScope public class xxController{ @value("${配置前缀.配置属性}") private String 对象 } 通过@ConfigurationProperties 我们可以创建一个类,在类里面添加一个属性,然后把配置的值注入进属性中java @Component @Data @ConfigurationProperties("配置前缀") public class 自定义类{ private String 配置属性 } 使用时我们仅需get数据即可``` java @Autowrite private 自定义类 对象public void method(){ 对象.get属性 } ```多环境共享在微服务启动时,nacos会读取多个服务 [spring.application.name]-[spring.profiles.active].[spring.cloud.nacos.config.file-extension],例如userservice-dev.properties [spring.application.name].[spring.cloud.nacos.config.file-extension],例如userservice.properties 因此,无论环境怎么变化,[spring.application.name].[spring.cloud.nacos.config.file-extension]一定会被加载,可以把共用的配置写在这个文件需要注意的是,多环境配置有优先级,若属性相同,以优先级高的为准Nacos集群企业使用的时候,Nacos不可能在一个服务器里,因此需要配置集群 搭建Nacos数据库,用来存储配置和账户等信息 配置Nacos,修改端口,数据库信息等等 在conf文件里将cluster.conf.example重命名为cluster.conf,并修改里面的值 直接启动Nacos,用命令startup.cmd就是集群启动 配置nignx.conf Gateway网关网关是一个服务器里的不同服务的负载均衡,流量监控的管理中心比较老的网关内部转发是Zuul,而本次学习使用最新的网关技术Gateway安装创建一个Model工程,导入依赖xml <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-gateway</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency>这里安装有一个注意事项,gateway和springWeb有冲突,若父工程由springWeb,需要重写springWeb为testxml <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> <scope>test</scope> </dependency>配置yaml server: port: 10010 spring: application: name: gateway cloud: nacos: config: server-addr: localhost:8848 gateway: routes: - id: user_service #唯一id uri: lb://userservice #ip地址,也可以直接写url predicates: - Path=/xxx/** #接口位置 filters: - Path=/xxx/** #路由过滤,请求处理或```properties应用服务 WEB 访问端口server.port=10010应用名称spring.application.name= gatewaynacos ip和地址spring.cloud.nacos.server-addr=localhost:8848路由的id，保持唯一即可spring.cloud.gateway.routes[0].id = userservice提供服务的路由地址spring.cloud.gateway.routes[0].uri=lb://userservice断言，路径相匹配的进行路由spring.cloud.gateway.routes[0].predicates[0]=Path=/user/** ```uri部分直接写url就等于使用了微服务网关 uri部分写lb://nacos注册Ip,有负载均衡效果,等于使用了api网关断言类型过滤器过滤器可以对请求和响应做处理```properties路由过滤器,作用当前路由spring.cloud.gateway.routes[0].filters[0]=AddRequestHeader=键,值默认过滤器,作用所有路由spring.cloud.gateway.default-filters=AddRequestHeader=键,值 ```自定义全局过滤器 实现GlobalFilter 添加@Order注解或实现Ordered接口 编写处理逻辑 ```java @Component @Order(int类型filter顺序) public class demoFilter implements GlobalFilter { @Override public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) {// //放行 // return chain.filter(exchange); //设置状态码并拦截 exchange.getResponse().setStatusCode(HttpStatus.MULTI_STATUS); return exchange.getResponse().setComplete(); } } ```过滤器顺序 每个过滤器都必须指定一个order值,order值越小,优先级越高,执行顺序越靠前 GlobalFilter通过实现@Order或Ordered接口来指定order值,具体值由我们自己决定 路由过滤器和默认过滤器的order由spring决定,默认按声明顺序从1开始 当order值一样时,默认过滤器>路由过滤器>全局自定义过滤器 跨域``` yaml spring: cloud: gateway: # 。。。 globalcors: # 全局的跨域处理 add-to-simple-url-handler-mapping: true # 解决options请求被拦截问题 corsConfigurations: '[/*]': allowedOrigins: # 允许哪些网站的跨域请求 - "http://localhost:8090" allowedMethods: # 允许的跨域ajax的请求方式 - "GET" - "POST" - "DELETE" - "PUT" - "OPTIONS" allowedHeaders: "" # 允许在请求中携带的头信息 allowCredentials: true # 是否允许携带cookie maxAge: 360000 # 这次跨域检测的有效期```
认识Elasitcsearch一个开源的分布式搜索引擎,可以用来实现搜索、日志统计、分析、系统监控等Elesticsearch核心技术就是倒排索引倒排索引和正向索引正向索引传统数据库如Mysql就是使用正向索引 根据ID搜索数据,数据库就会使用B+树来进行快速查询但如果是查询词条,就没有ID,而是逐行查询,这样一旦数据过多就会导致查询速度过慢倒向索引倒向索引弥补传统数据库正向索引中,查询词条过慢的技术根据ID,把正向索引数据以文档id和词条形式保存至倒向索引内,用户查询时根据词条使用B+树查询根据词条可以得到对应的ID,这样就可以根据ID直接查询数据库,避免数据库逐行查询Mysql和ESmysql和Es概念区别mysql和Es的关系mysql和Es是互补关系,一个项目中往往同时存在Mysql擅长事务操作,负责数据增删改 Es擅长海量数据搜索,负责查询安装安装Estex docker pull elasticsearch:7.17.7tex docker network create es-net```sh docker run -d \ --name elasticsearch \ -e ES_JAVA_OPTS="-Xms64m -Xmx128m" \ -e "discovery.type=single-node" \ -v es-data:/usr/share/elasticsearch/data \ -v es-plugins:/usr/share/elasticsearch/plugins \ --privileged \ --network es-net \ -p 9200:9200 \ -p 9300:9300 \ elasticsearch:7.17.7docker run -d \ --name elasticsearch \ -e "discovery.type=single-node" \ -v es-data:/usr/share/elasticsearch/data \ -v es-plugins:/usr/share/elasticsearch/plugins \ --privileged \ --network es-net \ -p 9200:9200 \ -p 9300:9300 \ elasticsearch:7.17.7 ```安装kibanakibana是一个Es可视化界面tex docker pull kibana:7.12.1```tex docker run -d \ --name kibana \ -e ELASTICSEARCH_HOSTS=http://elasticsearch:9200 \ --network es-net \ -p 5601:5601 \ kibana:7.12.1```安装分词器官方默认的分词方式对中文效果不好,因此需要额外下载一个分词器进入github下载对应的中文分词 https://github.com/medcl/elasticsearch-analysis-ik查找Es设置目录 tex docker volume inspect es-plugins将下载的压缩包解压,重命名为ik,并放入_data目录json POST /_analyze { "analyzer": "ik_smart", "text": "黄大旭真的帅" }分词器分词器拓展和停用ik分词器有ik_smart和ik_max_word通常情况下,分词器没法对当前热词进行分词,需要我们手动拓展,或者当我们需要手动设置敏感和无关词汇停用时,也需要手动配置在ik分词器目录中的config中的IKAnalyzer.cfg.xml定义自己的分词器DSL语句操作mapping属性mapping是对索引库中文档的约束,常见的索引有 type :字段数据类型,常见的简单类型有: 字符串:text(可分词的文本)、keyword(精准值,如:品牌、国家、Ip地址) 数值:long、interger、short、byte、double、float 布尔:boolean 日期:date 对象:object 坐标:geo_point(经纬度)、geo_shape(区域) 自动补全completion index:是否创建倒排索引,默认为ture(无倒排索引不会被查询) analyzer:使用哪种分词器 properties:该字段的子字段 创建索引库Es使用的是DQL语法,这个和json相似,也没必要记住,直接CVjson PUT /索引库名称 { "mappings": { "properties": { "字段名":{ "type": "text", "analyzer": "ik_smart" }, "字段名2":{ "type": "keyword", "index": false }, "字段名3":{ "type": "object", "properties": { "子字段": { "type": "keyword" } } }, // ...略 } } }创建拼音分词器索引库拼音分词注意事项 若想要索引库按拼音来分词,应先用中文分词,分出来的词再用拼音分,因而需要使用analysis 官方拼音分词有一些默认设置需要改动,比如我们不需要一个字一个字的分成拼音,需要再filter内改动,下面已经改好了 由于使用中文转拼音分词,所以当用户搜索中文时,可能会搜索到同音的数据,为避免这种情况,可以用search_analyzer设置搜索时只用中文分词,这样当用户搜索中文不会分词直接找索引库内的中文字段,当用户搜索拼音时直接找索引库内的拼音字段 json PUT /hotel { "settings": { "analysis": { "analyzer": { "text_anlyzer": { "tokenizer": "ik_max_word", "filter": "py" } }, "filter": { "py": { "type": "pinyin", "keep_full_pinyin": false, "keep_joined_full_pinyin": true, "keep_original": true, "limit_first_letter_length": 16, "remove_duplicated_term": true, "none_chinese_pinyin_tokenize": false } } } }, "mappings": { "properties": { "字段名":{ "type": "text", "analyzer": "text_anlyzer", "search_analyzer": "ik_smart" }, "字段名2":{ "type": "keyword", "index": false } } } }创建自动补全索引库当用户需要搜索A字段如商品,可以同个搜索B字段找到A字段自动补全注意事项: 自动补全需要额外定义一个字段(建议suggesion),类型为completion 用户搜索信息时找矿泉水,通常搜索的不仅仅是一个字段的,比如搜索哇哈哈或昆仑山,这里有两个字段品牌和生场地,因而我们希望用户搜索品牌也能找到商品,搜索地名也能找到商品,需要我们存储在suggesion字段里的内容为数组 从设计角度思考,一般用户搜索的自动补全都是某个品牌或地名,然后用户再自己手动输入详细数据,因此自动补全的suggestion的数组内容也都是keyword这些数据,不是可拆分的详细数据 查询索引库json GET /索引库名删除索引库json DELETE /索引库名修改索引库Es不支持修改已存在的字段,但是可以添加字段json PUT /索引库名/_mapping { "properties": { "新字段名":{ "type": "integer" } } }新增文档json POST /索引库名/_doc/文档id(主键id) { "字段1": "值1", "字段2": ["值2","值3"], "字段3": { "子属性1": "值4", "子属性2": "值5" }, // ... }删除文档json DELETE /{索引库名}/_doc/id值查询单个文档json GET /{索引库名}/_doc/id值修改文档全量修改(覆盖)文档全量修改是先删除再创建,执行了2次操作,而且原有数据会丢失json PUT /{索引库名}/_doc/文档id { "字段1": "值1", "字段2": "值2", // ... 略 }增量修改(合并)文档json POST /{索引库名}/_update/文档id { "doc": { "字段名": "新的值", } }合并字段将两个字段合并到一个字段里,并用分词器分开比如用户想在百度上搜索淘宝(网站名),但是可以通过搜索阿里巴巴(公司名)搜索到json PUT /索引库名称 { "mappings": { "properties": { "字段名":{ "type": "text", "analyzer": "ik_smart", "copy_to":"字段名3" }, "字段名2":{ "type": "keyword", "index": false, "copy_to":"字段名3" }, "字段名3":{ "type": "text", "analyzer": "ik_max_word", } } } }查询字段查询的基本语法json GET/索引库名称/_search { "query":{"查询类型":{"查询条件":值}} }查询所有查询所有数据,一般测试用 match_all json GET/索引库名称/_search { "query":{"match_all":{}} }全文检索利用分词器对用户输入的值进行分词,再去倒排索引库去匹配 math_query json GET/索引库名称/_search { "query":{ "math_query":{"字段":"搜索值"} } } multi_match_query 可以同时搜索多个字段,效果和all一样,不过由于多字段搜索,性能不如将数据copy到all里直接单字段查json GET/索引库名称/_search { "query":{ "multi_match_query":{ "query":"搜索值","fields":["字段1","字段2"] } } }精准查询根据精准词条查询,一般查询不可分词的数据 ids range 范围查询,一般查询数字,日期范围json GET/索引库名称/_search { "query":{ "range":{ "字段":{ "gte":大于等于>=, "gt":大于>, "lte":小于等于<=, "lt":小于< } } } } term 精准值查询,一般查询keyword,id,数值json GET/索引库名称/_search { "query":{"term":{"字段":"搜索值"}} }地理查询根据经纬度查询 geo_distance 查询一个圆,这里圆心格式一定要是"lat,lon"json GET/索引库名称/_search { "query":{ "geo_distance":{ "distance":xxkm, "字段":"lat,lon" } } } geo_bounding_box 查询一个矩形范围json GET/索引库名称/_search { "query":{ "geo_bounding_box":{ "字段":{ "top_left":{ "lat":latitude维度, "loc":longitude经度 }, "bottom_right":{ "lat":latitude维度, "loc":longitude经度 } } } } }复合查询复合查询,将上述的查询条件组合起来一起查询 bool bool可以用逻辑表达式查询,查询方式有: must,相当于与,参与相关性打分 should,相当于或,参与相关性打分 must_not,相当于否,不参与相关性打分 filter,相当于与,不参与相关性打分 json GET/索引库名称/_search { "bool":{ "查询方式":[ {"查询类型":{"查询条件":值}}, ... ], "查询方式":[ {"查询类型":{"查询条件":值}}, ... ], .... } } function_score 默认的查询是由相关性算法计算得分来排序,但是我们可以人为的修改得分从而使我们希望的数据得分更高,优先展示 json GET/索引库名称/_search { "query":{ "function_score":{ "query":{"查询类型":{"查询条件":"条件值"}}, "functions":{ "filter":{"查询类型":{"查询条件":"条件值"}}, "算分函数":值 }, boost_model:"加权模式" } } }排序Es默认使用相关性打分排序,如果指定其他排序,则不使用相关性打分 排序方式分为desc和asc 普通排序 json GET/索引库名称/_search { "query":{"查询类型":{"查询条件":值}}, "sort":{"字段":"排序方式"} } 地理坐标排序 地理坐标排序会返回一个sort,里面是用户与数据的距离json GET/索引库名称/_search { "query":{"查询类型":{"查询条件":值}}, "sort":{ "geo_distance":{ "字段":"用户lat,用户lon", "order":"排序方式", "unit":"返回单位(km)" } } }分页 from+size 优点:支持随机翻页缺点:深度分页问题,上限查询1万条场景:百度、京东、谷歌搜索都是只能翻70多页,每页10条数据json GET/索引库名称/_search { "query":{"查询类型":{"查询条件":值}}, "from":"第n个数据开始,不是页!!默认为0", "size":"往后n条数据,默认为10" } after search 优点:没有查询上限缺点:只能向后逐页查询场景:贴吧手机向下滚动查询 scroll(官方已放弃) 优点:没有查询上限缺点:占用内存场景:海量数据迁移高亮用户搜索的数据,我们需要将其高亮,而前端无法分析和分词用户搜索的数据,需要Es来对搜索数据打上标签默认情况下,搜索字段必须与高亮字段保持一致json GET/索引库名称/_search { "query":{"查询类型":{"match":{"搜索字段":"值"}}}, "highlight":{ "fildes":{ "高亮字段":{ "pre_tags":"<标签,推荐em>", "post_tags":"" } } } }但是多数情况下,我们搜索都是搜索all合并字段,然后只希望合并的某个字段高亮,这时候需要关闭高亮搜索匹配 json GET/索引库名称/_search { "query":{"查询类型":{"match":{"all":"值"}}}, "highlight":{ "fildes":{ "高亮字段":{ "pre_tags":"", "post_tags":"", "require_filed_match":"false" } } } }聚合分词字段聚合无意义,不可分词的字段才能被聚合聚合可以实现对文档数据的统计、分析和运算.聚合常见的三类有:桶(Buke)聚合用来对文档进行分组的 TermAggreation:按照文档分组(类似sql的group by) json GET/索引库名称/_search { "size":0, // 设置size为0,结果不包含文档,只有聚合结果 "aggs":{ "自定义聚合名称":{ "term":{ "filed":"字段", "size":20, //可选,设置显示的聚合数量 "order":{ "_count":"排序方式"//可选,按照桶大小排序 } } } } }​ 如果数据量较大,聚合的数目就会变多,这时候需要限制聚合数目json GET/索引库名称/_search { "query":{"range":{"字段":{"gte":大于等于>=}}}, "size":0, // 设置size为0,结果不包含文档,只有聚合结果 "aggs":{ "自定义聚合名称":{ "term":{ "filed":"字段", "size":20, //可选,设置显示的聚合数量 "order":{ "_count":"排序方式" //可选,按照桶大小排序 } } } } } Date Histogram:按照日期分组,如一周为一组或一月为一组 度量(Metric)聚合用来计算一些值,比如最大值最小值 Avg:求平均 Max:求最大 Min:求最小 Stats:可同时求最大最小 json GET/索引库名称/_search { "size":0, // 设置size为0,结果不包含文档,只有聚合结果 "aggs":{ "自定义聚合名称":{ "term":{ "filed":"字段", "size":20, //可选,设置显示的聚合数量 "order":{ "度量聚合名称.聚合函数":"排序方式" //可选,按照桶内的度量聚合结果排序 }, "aggs":{ "自定义度量聚合名称":{ "聚合函数":{ "filed":"字段" } } } } } } }管道(Pipeline)聚合把其他聚合的结果再聚合场景如果用户想要搜索每个品牌的平均分,这时候用桶聚合将品牌分类,最后度量聚合求分数平均自动补全自动补全只能搜索completion类型字段json GET/索引库名称/_search { "suggest":{ "自定义补全名称":{ "text":"关键字", "skip_duplicates": true,//去重 "size":10 //显示10个 } } }RestClient操作索引库Es官方提供了不同语言的客户端,用来操作Es,不需要用户手动发送DSL安装xml <properties> <elaticsearch.version>7.17.7</elaticsearch.version></properties> <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-high-level-client</artifactId> </dependency>初始化```java //使用前先创建连接 @BeforeEach void setUp() { this.client = new RestHighLevelClient(RestClient.builder( HttpHost.create("http://124.71.216.166:9200") )); }//使用client @Test void contextLoads() { System.out.println(client); }//关闭连接 @AfterEach void tearDown() throws IOException { this.client.close(); } ```创建索引库java @Test void create(){ // 1.创建request请求 CreateIndexRequest request = new CreateIndexRequest("索引库名称"); // 2.设置请求参数,String的DQL语法(不含PUT和库名,只要括号和括号内容),建议使用静态常量表示,XContentType表示语法格式 request.source("DQL语法", XContentType.JSON); // 3.发起请求,RequestOptions为请求头,不用管默认是DEFAULT client.indices().create(request, RequestOptions.DEFAULT); }删除索引库java @Test void delete() throws IOException { // 1.创建request请求 DeleteIndexRequest request = new DeleteIndexRequest("索引库名称"); // 2.发起请求 client.indices().delete(request, RequestOptions.DEFAULT); }查询索引库是否存在```java @Test void exists() throws IOException { // 1.创建request请求 GetIndexRequest request = new GetIndexRequest("索引库名称"); // 2.发起请求 boolean flag = client.indices().exists(request, RequestOptions.DEFAULT);System.out.println(flag ? "存在":"不存在"); } ```新增文档java @Test void add() throws IOException { // 1.创建request请求,文档ID要String IndexRequest request = new IndexRequest("索引库名称").id("文档ID"); // 2.设置请求参数,String的DQL语法(不含PUT和库名,只要括号和括号内容),建议使用静态常量表示,XContentType表示语法格式 request.source(Json.toString(新增对象), XContentType.JSON); // 3.发起请求,RequestOptions为请求头,不用管默认是DEFAULT client.index(request, RequestOptions.DEFAULT); }查询单个文档java @Test void get() throws IOException { // 1.创建request请求,文档ID要String GetRequest request = new GetRequest("索引库名称","文档ID"); // 2.发起请求,RequestOptions为请求头,不用管默认是DEFAULT GetResponse response = client.get(request, RequestOptions.DEFAULT); // 3.解析 String json = response.getResourceAsString(); }修改文档全量修改(覆盖文档)再次写入Id一样的文档,就会删除原有的,添加新的增量修改(合并)文档java @Test void update() throws IOException { // 1.创建request请求,文档ID要String UpdateRequest request = new UpdateRequest("索引库名称","文档ID"); // 2.准备参数 requset.doc( "age",18, "name","张三" ) // 3.发起请求,RequestOptions为请求头,不用管默认是DEFAULT client.update(request, RequestOptions.DEFAULT); }删除文档java @Test void delete() throws IOException { // 1.创建request请求,文档ID要String DeleteRequest request = new DeleteRequest("索引库名称","文档ID"); // 2.发起请求,RequestOptions为请求头,不用管默认是DEFAULT client.delete(request, RequestOptions.DEFAULT); }批量操作文档(适例新增)java @Test void bulk() throws IOException { // 1.创建request请求,文档ID要String BulkRequest request = new BulkRequest(); // 2. 添加批量请求,这里以批量新增为例 request.add(new IndexRequest("索引库名称").id("文档ID1").source(Json.toString(新增对象1), XContentType.JSON)); request.add(new IndexRequest("索引库名称").id("文档ID2").source(Json.toString(新增对象2), XContentType.JSON)); // 2.发起请求,RequestOptions为请求头,不用管默认是DEFAULT client.bulk(request, RequestOptions.DEFAULT); }查询字段查询字段基本语法java @Test void delete() throws IOException { // 1.创建request请求,文档ID要String SearchRequest request = new SearchRequest("索引库名称"); // 2.组织DSL参数 request.source().query(QueryBuilders.查询参数); // 3.发起请求,RequestOptions为请求头,不用管默认是DEFAULT SearchResponse response= client.search(request, RequestOptions.DEFAULT); // 4.处理请求 SearchHits searchHits = response.getHits(); // 4.1获取总条数 Long total = searchHits.getTotalHits().value; // 4.2获取数据 SearchHit[] hits = searchHits.getHits(); List<String> jsons = Arrays.stream(hits).map(hit -> hit.getSourceAsString()).collect(Collectors.toList()); }对应的查询参数有:查询所有java QueryBuilders.matchAllQuery()全文检索 math_query java QueryBuilders.matchQuery("字段","搜索值") multi_match_query 可以同时搜索多个字段,效果和all一样,不过由于多字段搜索,性能不如将数据copy到all里直接单字段查java QueryBuilders.multiMatchQuery("搜索值","字段1","字段2",...)精准查询 range 范围查询,一般查询数字,日期范围java QueryBuilders.termQuery("字段").gte(小于).lt(大于) term 精准值查询,一般查询keyword,id,数值java QueryBuilders.termQuery("字段","搜索值")地理查询复合查询 bool bool可以用逻辑表达式查询,查询方式有: must,相当于与,参与相关性打分 should,相当于或,参与相关性打分 must_not,相当于否,不参与相关性打分 filter,相当于与,不参与相关性打分 java // 添加布尔查询,不建议链式写,因为只能复合一条 BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery(); // 复合查询 boolQueryBuilder.查询方式(QueryBuilders.查询参数1); boolQueryBuilder.查询方式(QueryBuilders.查询参数2);排序Es默认使用相关性打分排序,如果指定其他排序,则不使用相关性打分 排序方式分为desc和asc 普通排序 java // 查询 request.source().query(QueryBuilders.查询参数); // 排序 request.source().sort(SortOrder.排序方式) 地理排序 地理坐标排序会返回一个sort,里面是用户与数据的距离java request.source.sort(SortBuilders.geoDistanceSort("字段",new GeoPoint("经度,纬度")).order(SortOrder.排序方式).unit(DistanceUnit.距离单位))分页 from+size ​ 优点:支持随机翻页 ​ 缺点:深度分页问题,上限查询1万条 ​ 场景:百度、京东、谷歌搜索都是只能翻70多页,每页10条数据java // 查询 request.source().query(QueryBuilders.查询参数); // 分页 request.source().form(第n个数据开始).size(往后n个数据) request.source().sort()高亮用户搜索的数据,我们需要将其高亮,而前端无法分析和分词用户搜索的数据,需要Es来对搜索数据打上标签默认情况下,搜索字段必须与高亮字段保持一致java request.source().highlighter(new HighlightBuilder().field("字段").preTags("默认").postTags("默认"));但是多数情况下,我们搜索都是搜索all合并字段,然后只希望合并的某个字段高亮,这时候需要关闭高亮搜索匹配java request.source().highlighter(new HighlightBuilder().field("搜索字段").requireFieldMatch(false).preTags("默认").postTags("默认"));java //高亮数据解析 List<String> jsons = Arrays.stream(hits).map( hit -> hit.getHighlightFields().get("高亮字段").getFragments()[0].toString()) .collect(Collectors.toList());聚合分词字段聚合无意义,不可分词的字段才能被聚合聚合可以实现对文档数据的统计、分析和运算.聚合常见的三类有:桶(Buke)聚合用来对文档进行分组的 TermAggreation:按照文档分组(类似sql的group by) java request.source().size(0); request.source().aggregation( AggregationBuilders .term("聚合名称") .filed("聚合字段") .size(int);//可选,聚合显示数目 );java //聚合解析 // 1. 解析结果 Aggregations aggregations = response.getAggregations(); // 2. 根据名称获取想要的聚合结果 Term brandTerms = aggregations.get("聚合名称"); // 3. 获取桶数据 List<? extends Terms.Bucket> buckets = brandTerms.getBuckets(); // 4. 遍历 for(Terms.Bucket bucket : buckets){ String result = bucket.getKeyAsString(); } Date Histogram:按照日期分组,如一周为一组或一月为一组 度量(Metric)聚合用来计算一些值,比如最大值最小值 Avg:求平均 Max:求最大 Min:求最小 Stats:可同时求最大最小 管道(Pipeline)聚合把其他聚合的结果再聚合场景自动补全java request.source().suggest(new SuggestBuilder().addSuggestion( "自定义补全名称", SuggestBuilders.completionSuggestion("字段") .prefix("关键字") .skipDuplicates(true)//去重 .size(10) ));java //解析 // 1.解析结果 Suggest suggest = response.getSuggest(); // 2.根据补全查询名称，获取补全结果 CompletionSuggestion suggestions = suggest.getSuggestion("自定义补全名称"); // 3.获取options List<CompletionSuggestion.Entry.Option> options = suggestions.getOptions(); // 4.遍历 List<String> list = new ArrayList<>(options.size()); for (CompletionSuggestion.Entry.Option option : options) { String text = option.getText().toString(); list.add(text); }相关性算法在Es中,分词查询后得到多个数据,我们需要对这些数据进行打分,让关联度高的数据优先展示TF(词条频率)比如一个数据叫虹桥如家酒店真不错,它可拆分为虹桥,如家,酒店和真不错四个数据 那么如果用户搜索虹桥如家,它可拆分为虹桥和如家两个数据其中,用户输入的虹桥的TF是0.25,如家的TF是0.25,那么总得分就是0.25+0.25=0.5 缺点 相关性搜索不够高如果另一个数据叫如家酒店,那么用户输入的虹桥的TF是0,如家的TF是0.5,那么总得分就是0+0.5=0.5TF-IDF当前搜索最流行的算法 这个算法是在求频率的基础上加了权重比如有两个数据虹桥如家酒店真不错,它可拆分为虹桥,如家,酒店和真不错四个数据| 如家酒店,它可拆分为如家,酒店两个数据 那么如果用户搜索虹桥如家,它可拆分为虹桥和如家两个数据其中,用户输入的如家两个数据都有,它的IDF(权重)为0,而虹桥的IDF(权重)为0.3因此,数据一的得分是:0.25 * 0.3+0.25 * 0=0.075 数据二的得分是0 * 0.3 + 0.5 * 0 = 0BM25算法ES5以后使用的算法 我看不懂,我大为震撼 数据同步同步调用(Fegin) 优点:实现简单 缺点:慢 异步调用(MQ) 优点:快 缺点:依赖MQ可靠性 监听binlog(cannel) 优点:快,而且低耦合,存储sql的服务不需要额外操作 缺点:sql压力增加 集群单机的Es做数据存储,必然面对两个问题:海量存储数据和单点故障 海量存储问题:将索引库拆分成N个分片,存储到多个节点 单点故障问题:将不同分片在不同节点备份 搭建集群使用docker-compose一键部署多个镜像```yaml version: '2.2' services: es01: image: elasticsearch:版本 container_name: es01 environment: - node.name=es01 - cluster.name=es-docker-cluster - discovery.seed_hosts=es02IP,es03IP(相同服务器就写容器名) - cluster.initial_master_nodes=es01,es02,es03 - "ES_JAVA_OPTS=-Xms512m -Xmx512m"# 不填默认1g volumes: - data01:/usr/share/elasticsearch/data ports: - 9200:9200 networks: - elastic es02: image: elasticsearch:版本 container_name: es02 environment: - node.name=es02 - cluster.name=es-docker-cluster - discovery.seed_hosts=es01IP,es03IP(相同服务器就写容器名) - cluster.initial_master_nodes=es01,es02,es03 - "ES_JAVA_OPTS=-Xms512m -Xmx512m"# 不填默认1g volumes: - data02:/usr/share/elasticsearch/data ports: - 9201:9200 networks: - elastic es03: image: elasticsearch:版本 container_name: es03 environment: - node.name=es03 - cluster.name=es-docker-cluster - discovery.seed_hosts=es01IP,es02IP(相同服务器就写容器名) - cluster.initial_master_nodes=es01,es02,es03 - "ES_JAVA_OPTS=-Xms512m -Xmx512m"# 不填默认1g volumes: - data03:/usr/share/elasticsearch/data networks: - elastic ports: - 9202:9200 volumes: data01: driver: local data02: driver: local data03: driver: localnetworks: elastic: driver: bridge ```如果启动失败,可能是linux内存不足,可以放开内存限制(不安全) es运行需要修改一些linux系统权限，修改/etc/sysctl.conf文件sh vi /etc/sysctl.conf添加下面的内容：sh vm.max_map_count=262144然后执行命令，让配置生效：sh sysctl -p### 集群小工具比kibana好用!不用在服务器安装https://github.com/lmenezes/cerebro通过docker-compose启动集群：sh docker-compose up -d集群职责集群的每个节点都有对应的任务,默认情况下每个节点都是身兼数职而一个健壮良好的Es集群应该是这样集群脑裂问题早期Es有一个问题,就是如果主节点和备选主节点网络断开,备选主节点就会选出一个新的主节点,这时候网络恢复,就出行了两个主节点,这就是脑裂问题ES7针对脑裂问题,设计了一个选举机制,超过(eligible节点数+1)/2的节点才有资格去选举主节点,这样就避免了脑裂问题集群故障转移Es集群会保证每个分片至少有两份在集群里,如果某节点宕机,那么其他节点会将它的数据和备份都拿来集群分布式存储分布式存储会根据文档Id进行哈希运算 routing是文档Id,number_of_shards为总data节点数 存储步骤:集群分布式查询分布式查询就简单了,所有节点都查,然后合并
介绍在学习Sentinel前,我们需要了解Sentinel能解决什么问题面对的问题1.雪崩问题当一台服务宕机时,其关联的其他服务会一直等待响应从而宕机,因此照成雪崩 超时处理:设定请求超时时间,超过时间未响应则返回错误信息,不是无休止等待 舱壁模式:限定每个业务的线程数,避免因无休止等待照成资源堵塞而宕机 熔断降级:由断路器统计业务异常比例,达到阈值则拦截该业务一切请求 2.流量暴增服务请求QPS流量突然暴增,导致服务压力过大而宕机 流量控制,限制业务访问的QPS 同类保护技术对比Sentinel是由阿里巴巴开发的,一种相对成熟的服务器保护技术安装安装Sentinel管理页面在https://github.com/alibaba/Sentinel下载最新版Sentinel管理页面控制台输入shell java -jar sentinel-dashboard-1.8.6.jar -Dserver.port=对应端口进入页面:localhost:对应端口,默认账号密码为sentinel,若需要修改其他数据,课参考此表微服务整合Sentinel引入依赖```xml com.alibaba.cloud spring-cloud-starter-alibaba-sentinel ```配置application.properties```properties配置sentinelspring.cloud.sentinel.transport.dashboard=localhost:8090 ```访问业务的任意端口,触发sentinel监控指定监控资源sentinel默认只检测Controler,若需要指定检测资源,需要手动添加注解java @Service @SentinelResource("资源名") public void xxx(){ }properties spring.cloud.sentinel.web-context-unify = false整合Feginsentinel监控默认内部资源,若需要监控其连接状况,需要自定义 1. 修改配置properties fegin.sentinel.enable = ture 编写异常处理 java public class xxxClientFallbackFactory implements FallbackFatory<Fegin连接类>{ @Override public Fegin连接类 create(Throwable throwable){ return new Fegin连接类{ @Override public 返回类 Fegin方法 (){ //Fegin连接对应的方法,使用Sentinel隔离和熔断规则 return 返回需要的类型; } } } }3.Fegin连接类指定规则java @FeignClient(value="/xx",FallbackFatory=xxxClientFallbackFactory) public interface Fegin连接类{ @GetMapping("/xx") 返回类 Fegin方法(); }流控模式流控模式对服务的并发情况进行监控,分为QPS(流量监控)和线程监控(线程隔离) 直接:对当前资源限流,超过QPS/线程不予访问 关联:对当前资源限流,超过QPS/线程关联资源不予访问 链路:对当前资源(service/mapper)限流,超过QPS/线程不予对应链路的(controller)访问 QPS(流量监控)流量模式可针对服务接收的请求进行监控热点模式热点模式属于QPS(流量监控模式)的分支,可以对Restful(/{id})请求进行监控,可单独给该请求id设置独立的限流规则线程监控(线程隔离)当前技术下,线程隔离主要有一下两种 线程池隔离为服务开启指定线程,用于不用都一直占用内存优点:异步,不用等待线程开启缺点:服务器开销大 信号量隔离(sentinel使用)使用计数器计算服务线程数,线程数到阈值不开启线程优点:轻量缺点:不支持异常,需要等待线程开启 流控效果 快速失败:超过QPS/线程不予访问 warm up:服务刚开启时,阈值逐渐上升 排队等待:请求进入队列等待,计算每个请求的等待时间,超过阈值不予访问 熔断模式虽然限流可以防止高并发而导致服务崩坏,但服务还会因为其他原因而崩坏,为避免雪崩,需要设置熔断降级熔断降级会判断服务的异常比例(close模式),若高于阈值则关闭连接n秒(open模式),时间到尝试再一次放行请求查看是否异常(Half-open模式) 慢调用:熔断器计算请求连接时间,超出RT秒比例高于阈值则熔断 异常比例:熔断器计算请求抛出异常比例 异常数:熔断器计算请求抛出异常个数 授权模式只有携带特定数据才能访问服务,该数据可以是请求头数据、参数等信息通常情况下,我们希望用户从网关进入服务,但如果用户通过特殊方法获取服务ip端口,网关就不能进行监控作用.因此需要网关发送特点请求头,只有请求头携带特定的数据才允许访问Sentinel是通过RequestOrignParser接口来判断数据是否授权,默认该接口返回default因此只要实现此接口,将网关携带的请求头数据拿来判断是否授权即可java @Component public class HeaderOrginParser implements RequestOriginParser { @Override public String parseOrigin(HttpServletRequest httpServletRequest) { String orgin = httpServletRequest.getHeader("orgin"); if (StringUtils.isEmpty(orgin)){ return "black"; } return orgin; } }gate配置发送请求头properties spring.cloud.gateway.default-filters=AddRequestHeader=orgin,gateway 流控应用:携带的数据 白名单:数据运行通行 黑名单:数据不允通行 自定义异常当Sentinel发生流控、熔断、授权或热点拦截时,都会抛出异常流控异常并响应,而我们可自定义异常响应Sentinel是通过BlockExceptionHandel接口来处理异常,我们需要实现此接口返回我们自定义的异常格式BlockException有很多子类```java @Component public class SentinelExceptionHandler implements BlockExceptionHandler { @Override public void handle(HttpServletRequest request, HttpServletResponse response, BlockException e) throws Exception { String msg = "未知异常"; int status = 429; if (e instanceof FlowException) { msg = "请求被限流了"; } else if (e instanceof ParamFlowException) { msg = "请求被热点参数限流"; } else if (e instanceof DegradeException) { msg = "请求被降级了"; } else if (e instanceof AuthorityException) { msg = "没有权限访问"; status = 401; } response.setContentType("application/json;charset=utf-8"); response.setStatus(status); response.getWriter().println("{\"msg\": " + msg + ", \"status\": " + status + "}"); } } ```持久化Sentinel有三种配置管理模式: 原始模式:保存在内存 pull模式:保存在本地或数据库,定时读取 push模式:保存在nacos,实时获取 sentinel模式将配置保存在内存,一旦重启,所有配置将会失效push模式: 引入依赖 xml <dependency> <groupId>com.alibaba.csp</groupId> <artifactId>sentinel-datasource-nacos</artifactId> </dependency> 配置nacos地址 yaml spring: sentinel: datasource: flow: nacos: server-addr: localhost:8848 # nacos地址 dataId: orderservice-flow-rules groupId: SENTINEL_GROUP rule-type: flow # 还可以是：degrade(熔断)、authority(授权)、param-flow(热点) degrade: nacos: server-addr: localhost:8848 # nacos地址 dataId: orderservice-flow-rules groupId: SENTINEL_GROUP rule-type: degrade # 还可以是：flow(限流)、authority、param-flow
任务调度技术任务调度是特定业务场景下的定时任务处理，在分布式架构下，分布式调度框架的设计显得尤为重要。这里简要介绍了两种常用的分布式调度框架Quartz和xxl-job的特性、基本架构和参数配置，以加深了解。Spring Schedulespring自带一种基于注解的任务调度在application中添加@EnableScheduling注解```java @SpringBootApplication @EnableScheduling //开启定时任务功能 public class SpringbootTaskApplication {public static void main(String[] args) { SpringApplication.run(SpringbootTaskApplication.class, args); } }```在定时任务类上添加规则和方法,使用cron表达式```java @Slf4j @Component public class MyJob {//@Scheduled(cron = "0/${mytask.abc:15} * * * * ?") 在application.yml配置中添加参数可以通过配置方式指定规则 mytask.abc = 10 ; @Scheduled(cron = "0,15,30,45 * * * * ?") public void hello(){ log.info("say hello"); } } ```Quartz介绍这是一个作业调度的开源项目,一般可用和spring结合使用,在实际开发中可用用 Quartz做一个定时任务,比如每小时执行一次,或每个月上午10点执行一次,每个月下午5点执行一次安装```xml org.springframework.boot spring-boot-starter-quartz ```使用 编写任务信息 编写任务处理规则 配置调度中心(启动时开启任务调度) 首先,先编写job类```java public class Job1 implements Job { private final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");private void before() { System.out.println("任务1：开始执行-" + dateFormat.format(new Date())); } @Override public void execute(JobExecutionContext arg0) throws JobExecutionException { before(); System.out.println("任务1：业务逻辑。。。"); after(); } private void after() { System.out.println("任务1：执行结束"); System.out.println(); } ```再通过工具类配置规则```java @Component public class QuartzSchedulerManager { @Autowired private Scheduler scheduler;// 开始执行定时器 public void startJob() throws SchedulerException { startJob1(scheduler); startJob2(scheduler); scheduler.start(); } // 获取Job信息 public String getJobInfo(String name, String group) throws SchedulerException { TriggerKey triggerKey = new TriggerKey(name, group); CronTrigger cronTrigger = (CronTrigger) scheduler.getTrigger(triggerKey); return String.format("time:%s,state:%s", cronTrigger.getCronExpression(), scheduler.getTriggerState(triggerKey).name()); } // 修改某个任务的执行时间 public boolean modifyJob(String name, String group, String time) throws SchedulerException { Date date = null; TriggerKey triggerKey = new TriggerKey(name, group); CronTrigger cronTrigger = (CronTrigger) scheduler.getTrigger(triggerKey); String oldTime = cronTrigger.getCronExpression(); if (!oldTime.equalsIgnoreCase(time)) { CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(time); CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(name, group) .withSchedule(cronScheduleBuilder).build(); date = scheduler.rescheduleJob(triggerKey, trigger); } return date != null; } // 暂停所有任务 public void pauseAllJob() throws SchedulerException { scheduler.pauseAll(); } // 暂停某个任务 public void pauseJob(String name, String group) throws SchedulerException { JobKey jobKey = new JobKey(name, group); JobDetail jobDetail = scheduler.getJobDetail(jobKey); if (jobDetail == null) return; scheduler.pauseJob(jobKey); } // 恢复所有任务 public void resumeAllJob() throws SchedulerException { scheduler.resumeAll(); } // 恢复某个任务 public void resumeJob(String name, String group) throws SchedulerException { JobKey jobKey = new JobKey(name, group); JobDetail jobDetail = scheduler.getJobDetail(jobKey); if (jobDetail == null) return; scheduler.resumeJob(jobKey); } // 删除某个任务 public void deleteJob(String name, String group) throws SchedulerException { JobKey jobKey = new JobKey(name, group); JobDetail jobDetail = scheduler.getJobDetail(jobKey); if (jobDetail == null) return; scheduler.deleteJob(jobKey); } // 启动任务1 private void startJob1(Scheduler scheduler) throws SchedulerException { // 通过JobBuilder构建JobDetail实例，JobDetail规定其job只能是实现Job接口的实例 JobDetail jobDetail = JobBuilder.newJob(Job1.class).withIdentity("job1", "group1").build(); // 基于表达式构建触发器 CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule("0/5 * * * * ?"); // CronTrigger表达式触发器 继承于Trigger。TriggerBuilder 用于构建触发器实例 CronTrigger cronTrigger = TriggerBuilder.newTrigger().withIdentity("job1", "group1") .withSchedule(cronScheduleBuilder).build(); scheduler.scheduleJob(jobDetail, cronTrigger); } // 启动任务2 private void startJob2(Scheduler scheduler) throws SchedulerException { // 通过JobBuilder构建JobDetail实例，JobDetail规定其job只能是实现Job接口的实例 JobDetail jobDetail = JobBuilder.newJob(Job2.class).withIdentity("job2", "group2").build(); // 基于表达式构建触发器 CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule("0/5 * * * * ?"); // CronTrigger表达式触发器 继承于Trigger。TriggerBuilder 用于构建触发器实例 CronTrigger cronTrigger = TriggerBuilder.newTrigger().withIdentity("job2", "group2") .withSchedule(cronScheduleBuilder).build(); scheduler.scheduleJob(jobDetail, cronTrigger); } } ```若希望服务启动时运行,写以下配置类```java @Configuration public class QuartzListener implements ApplicationListener { @Autowired private QuartzSchedulerManager quartzSchedulerManager;// 初始启动quartz @Override public void onApplicationEvent(ContextRefreshedEvent event) { try { quartzSchedulerManager.startJob(); System.out.println("任务已经启动..."); } catch (SchedulerException e) { e.printStackTrace(); } } // 初始注入scheduler @Bean public Scheduler scheduler() throws SchedulerException{ SchedulerFactory schedulerFactoryBean = new StdSchedulerFactory(); return schedulerFactoryBean.getScheduler(); } } ```若希望通过api启动任务类,在controler里执行以下api```java @RestController @RequestMapping("") public class QuartzController { @Autowired private QuartzSchedulerManager quartzSchedulerManager;// @Description: 获取定时器信息 @GetMapping("/info") public String getQuartzJob(String name, String group) { String info = null; try { info = quartzSchedulerManager.getJobInfo(name, group); } catch (SchedulerException e) { e.printStackTrace(); } return info; } // @Description: 修改定时器的 执行时间 @PostMapping("/modify") public boolean modifyQuartzJob(String name, String group, String time) { boolean flag = true; if (!CronExpression.isValidExpression(time)) { throw new RuntimeException("非法的cron表达式"); } try { flag = quartzSchedulerManager.modifyJob(name, group, time); } catch (SchedulerException e) { e.printStackTrace(); } return flag; } // @Description: 启动所有定时器 @PostMapping("/start") public void startQuartzJob() { try { quartzSchedulerManager.startJob(); } catch (SchedulerException e) { e.printStackTrace(); } } // @Description: 暂停指定 定时器 @PostMapping(value = "/pause") public void pauseQuartzJob(String name, String group) { try { quartzSchedulerManager.pauseJob(name, group); } catch (SchedulerException e) { e.printStackTrace(); } } // 暂停所有定时器 @PostMapping(value = "/pauseAll") public void pauseAllQuartzJob() { try { quartzSchedulerManager.pauseAllJob(); } catch (SchedulerException e) { e.printStackTrace(); } } // 删除指定定时器 @PostMapping(value = "/delete") public void deleteJob(String name, String group) { try { quartzSchedulerManager.deleteJob(name, group); } catch (SchedulerException e) { e.printStackTrace(); } } } ```XXL-JOB介绍xxl-job是基于分布式集群的任务调度模块 单机情况下,一个任务调度中心可以发起一种任务调度 但在集群情况下,多个任务调度中心就会重复发一种任务调度 如我们希望在每个月1号清空积分,但用了集群分布式,有了多个调度中心,那么每个调度中心都会清空积分,这样任务就重复了,我们希望只有一个任务中心发起任务调度即可,而xxl-job就是专门解决分布式集群任务调度的开源框架 执行服务:编写任务信息 调度服务:编写任务规则 执行服务:配置调度中心(启动时开启任务调度) cron表达式对于cron表达式,如果自己写的话会比较麻烦,可以使用网络上的cron表达式在线 生成工具https://cron.qqe2.com/
多级缓存概述多级缓存是一种设计思路,充分利用请求的每个环境进行缓存,减少Tomcat的压力 一级缓存:浏览器客户端缓存,主要由cookie或LocalStroage 二级缓存:Nginx本地缓存 三级缓存:Redis缓存,主要由Nginx用Lua发起向Redis的询问 四级缓存:JVM缓存,进程缓存 五级缓存:本地数据库 JVM缓存JVM是一种基于本地进程的缓存,相较于Redis其使用本地缓存,没有网络开销(Redis可能使用其他服务器)速度更快,但是不能数据共享.常见的JVM缓存就是以静态变量的形式进行缓存,线程未结束,变量就不会被释放.Spring官方推荐使用Caffeine,其内部就是使用静态Map创建Caffeinejava //创建构造器 Cache<String,String> cache = Caffeine.newBuilder().build();创建构造器时可添加其他功能 initialCapacity(long):存储初始化大小 maximumSize(long):存储最大大小 expireAfterWrite(Duration.ofSeconds(long)):存储存long秒后释放 一般在Config里创建Caffeinejava @Configuration public class CaffeineConfig { @Bean public Cache<String,String> XXCache(){ Cache<String,String> cache = Caffeine.newBuilder() .initialCapacity(1) .maximumSize(1) .expireAfterWrite(Duration.ofSeconds(10)) .build(); return cache; } }使用Caffeinejava //存数据 cache.put("键","值");java //取数据,若无数据返回Null cache.getIfPresent("键"); //取数据,若无数据执行方法 cache.get("键",key->{ return ""; });Nginx使用RedisNginx配置修改使用前需要导入OpenResty```nginxuser nobody;worker_processes 1; error_log logs/error.log;events { worker_connections 1024; }http { include mime.types; default_type application/octet-stream; sendfile on; keepalive_timeout 65; #导入lua模块 lua_package_path "user/local/openrestry/lualib/?.lua;;"; #导入c模块 lua_package_cpath "user/local/openrestry/lualib/?.so;;"; #设置集群 upstream tomcat-cluster{ server tomcatIp1:端口; server tomcatIp2:端口; } server { #监听端口和域名 listen 8081; server_name localhost; #映射位置 location / { root html; index index.html index.htm; } location /item{ proxy_pass http://tomcat-cluster; } #设置某接口监听由lua脚本控制 location ~ /api/item/(\d+){ # 默认响应类型 default_type application/json; # 响应结果由lua/item.lua文件决定 content_by_lua_file lua/item.lua } error_page 500 502 503 504 /50x.html; location = /50x.html { root html; } } } ```导入common.lua```lua -- 导入redis local redis = require('resty.redis') -- 初始化redis local red = redis:new() red:set_timeouts(1000,1000,1000)-- 关闭redis连接的工具方法，其实是放入连接池 local function close_redis(red) local pool_max_idle_time = 10000 -- 连接的空闲时间，单位是毫秒 local pool_size = 100 --连接池大小 local ok, err = red:set_keepalive(pool_max_idle_time, pool_size) if not ok then ngx.log(ngx.ERR, "放入redis连接池失败: ", err) end end -- 查询redis的方法 ip和port是redis地址，key是查询的key local function read_redis(ip, port, key) -- 获取一个连接 local ok, err = red:connect(ip, port) if not ok then ngx.log(ngx.ERR, "连接redis失败 : ", err) return nil end -- 查询redis local resp, err = red:get(key) -- 查询失败处理 if not resp then ngx.log(ngx.ERR, "查询Redis失败: ", err, ", key = " , key) end --得到的数据为空处理 if resp == ngx.null then resp = nil ngx.log(ngx.ERR, "查询Redis数据为空, key = ", key) end close_redis(red) return resp end -- 封装函数，发送http请求，并解析响应 local function read_http(path, params) local resp = ngx.location.capture(path,{ method = ngx.HTTP_GET, args = params, }) if not resp then -- 记录错误信息，返回404 ngx.log(ngx.ERR, "http not found, path: ", path , ", args: ", args) ngx.exit(404) end return resp.body end -- 将方法导出 local _M = {read_http = read_http, read_redis = read_redis }return _M ```导入item.lua```lua -- 导入common库,获取发送http请求方法 local common = require('common') local read_http = common.read_http local read_redis = common.read_redis -- 导入json库 local cjson = require('cjson')-- 获取路径参数 local id = ngx.var[1]-- 封装查询函数,先查redis,后发http function read_data(key,path,params) -- 先查redis local resp = read_redis("127.0.0.1",6379,key) -- 判断查询结果,未空则发http if not resp then ngx.log(ngx.ERR,"redis查询失败,尝试查询http,key:",key) resp = read_http(path,params) end return resp end -- 查询 local itemJson = read_data("item.id"..id,"/item/"..id,nil) local itemStockJson = read_data("item.stock.id"..id,"/item/stock/"..id,nil) -- 组合json local item = cjson.decode(itemJson) local stock = cjson.decode(itemStockJson) item.stock = stock.stock item.sold = stock.sold-- 返回结果 ngx.say(cjson.encode(item)) ```Nginx本地缓存OpenResty为Nginx提供了shard dict的功能,在nginx的多个worker之间共享数据,实现缓存功能.Nginx配置修改在nginx.conf中添加配置```nginx开启本地缓存,名字为item_cache,大小为150mlua_shard_dict item_cache 150m ``````nginxuser nobody;worker_processes 1; error_log logs/error.log;events { worker_connections 1024; }http { include mime.types; default_type application/octet-stream; sendfile on; keepalive_timeout 65; #导入lua模块 lua_package_path "user/local/openrestry/lualib/?.lua;;"; #导入c模块 lua_package_cpath "user/local/openrestry/lualib/?.so;;"; #开启本地缓存,名字为item_cache,大小为150m lua_shard_dict item_cache 150m; #设置集群 upstream tomcat-cluster{ server tomcatIp1:端口; server tomcatIp2:端口; } server { #监听端口和域名 listen 8081; server_name localhost; #映射位置 location / { root html; index index.html index.htm; } location /item{ proxy_pass http://tomcat-cluster; } #设置某接口监听由lua脚本控制 location ~ /api/item/(\d+){ # 默认响应类型 default_type application/json; # 响应结果由lua/item.lua文件决定 content_by_lua_file lua/item.lua } error_page 500 502 503 504 /50x.html; location = /50x.html { root html; } } } ```本地缓存操作在lua中操作lua -- 获取本地缓存对象 local item_cache = ngx.shared.item_cache -- 存储,指定key,value,过期时间,单位s,默认0为永不过期 item_cache:set('key','value',1000) -- 读取 local val = item_cache:get('key')修改item.lua如在上文的item.lua中,可先查询本性再查询redis,最后发送http```lua -- 导入common库,获取发送http请求方法 local common = require('common') local read_http = common.read_http local read_redis = common.read_redis -- 导入json库 local cjson = require('cjson') -- 获取本地缓存对象 local item_cache = ngx.shared.item_cache-- 获取路径参数 local id = ngx.var[1]-- 封装查询函数,先查redis,后发http function read_data(key,expire,path,params) -- 查询本地缓存 local val = item_cache:get(key) if not val then -- 查redis ngx.log(ngx.ERR,"本性缓存查询失败,尝试查询redis,key:",key) val = read_redis("127.0.0.1",6379,key) -- 判断查询结果,未空则发http if not val then ngx.log(ngx.ERR,"redis查询失败,尝试查询http,key:",key) val = read_http(path,params) end end item_cache:set(key,val,expire) return val end-- 查询 local itemJson = read_data("item.id"..id,1800,"/item/"..id,nil) local itemStockJson = read_data("item.stock.id"..id,60,"/item/stock/"..id,nil) -- 组合json local item = cjson.decode(itemJson) local stock = cjson.decode(itemStockJson) item.stock = stock.stock item.sold = stock.sold-- 返回结果 ngx.say(cjson.encode(item)) ```缓存同步MQ同步通过MQ,当item服务更新时,发布更新消息,cache服务更新redis优点:快缺点:依赖MQ可靠性Cannel同步通过监听mysql的binlog,一旦触发增删改事件,则同步更新redis优点:快,低耦合(不用再item服务写代码)缺点:依赖数据库可靠性安装Cannel1.修改msql配置sh vi /tmp/mysql/conf/my.cnfini log-bin=/var/lib/mysql/mysql-bin binlog-do-db=heima配置解读： log-bin=/var/lib/mysql/mysql-bin：设置binary log文件的存放地址和文件名，叫做mysql-bin binlog-do-db=heima：指定对哪个database记录binary log events，这里记录heima这个库 ### 2.设置用户权限接下来添加一个仅用于数据同步的账户，出于安全考虑，这里仅提供对heima这个库的操作权限。mysql create user canal@'%' IDENTIFIED by 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%' identified by 'canal'; FLUSH PRIVILEGES;重启mysql容器即可docker restart mysql测试设置是否成功：在mysql控制台，或者Navicat中，输入命令：show master status;3.创建网络我们需要创建一个网络，将MySQL、Canal、MQ放到同一个Docker网络中：sh docker network create heima让mysql加入这个网络：sh docker network connect heima mysql4.安装Cannel网络上下载cannel,可以上传到虚拟机，然后通过命令导入：docker load -i canal.tar然后运行命令创建Canal容器：sh docker run -p 11111:11111 --name canal \ -e canal.destinations=heima \ -e canal.instance.master.address=mysql:3306 \ -e canal.instance.dbUsername=canal \ -e canal.instance.dbPassword=canal \ -e canal.instance.connectionCharset=UTF-8 \ -e canal.instance.tsdb.enable=true \ -e canal.instance.gtidon=false \ -e canal.instance.filter.regex=heima\\..* \ --network heima \ -d canal/canal-server:v1.1.5说明: -p 11111:11111：这是canal的默认监听端口 -e canal.instance.master.address=mysql:3306：数据库地址和端口，如果不知道mysql容器地址，可以通过docker inspect 容器id来查看 -e canal.instance.dbUsername=canal：数据库用户名 -e canal.instance.dbPassword=canal ：数据库密码 -e canal.instance.filter.regex=：要监听的表名称 表名称监听支持的语法：mysql 数据解析关注的表，Perl正则表达式. 多个正则之间以逗号(,)分隔，转义符需要双斜杠(\\) 常见例子： 1. 所有表：.* or .*\\..* 2. canal schema下所有表： canal\\..* 3. canal下的以canal打头的表：canal\\.canal.* 4. canal schema下的一张表：canal.test1 5. 多个规则组合使用然后以逗号隔开：canal\\..*,mysql.test1,mysql.test2使用Cannel导入依赖xml <dependency> <groupId>top.javatool</groupId> <artifactId>canal-spring-boot-starter</artifactId> <version>1.2.1-RELEASE</version> </dependency>编写配置yaml canal: destination: heima # canal的集群名字，要与安装canal时设置的名称一致 server: 192.168.150.101:11111 # canal服务地址修改实体类通过@Id、@Column、等注解完成Item与数据库表字段的映射：```java package com.heima.item.pojo;import com.baomidou.mybatisplus.annotation.IdType; import com.baomidou.mybatisplus.annotation.TableField; import com.baomidou.mybatisplus.annotation.TableId; import com.baomidou.mybatisplus.annotation.TableName; import lombok.Data; import org.springframework.data.annotation.Id; import org.springframework.data.annotation.Transient;import javax.persistence.Column; import java.util.Date;@Data @TableName("tb_item") public class Item { @TableId(type = IdType.AUTO) @Id private Long id;//商品id @Column(name = "name") private String name;//商品名称 private String title;//商品标题 private Long price;//价格（分） private String image;//商品图片 private String category;//分类名称 private String brand;//品牌名称 private String spec;//规格 private Integer status;//商品状态 1-正常，2-下架 private Date createTime;//创建时间 private Date updateTime;//更新时间 @TableField(exist = false) @Transient private Integer stock; @TableField(exist = false) @Transient private Integer sold; } ```编写监听类通过实现EntryHandler<T>接口编写监听器，监听Canal消息。注意两点： 实现类通过@CanalTable("tb_item")指定监听的表信息 EntryHandler的泛型是与表对应的实体类 ```java @CanalTable("tb_item") @Component public class ItemHandler implements EntryHandler {@Autowired private RedisHandler redisHandler; @Autowired private Cache<Long, Item> itemCache; //发生新增时触发事件 @Override public void insert(Item item) { // 写数据到JVM进程缓存 .... // 写数据到redis ... } //发生更新时触发事件 @Override public void update(Item before, Item after) { // 写数据到JVM进程缓存 ... // 写数据到redis ... } //发生删除时触发事件 @Override public void delete(Item item) { // 删除数据到JVM进程缓存 ... // 删除数据到redis ... } } ```
无论是密码登录还是第三方微信登录,我们都需要向前端返回一个token,这个过程就叫认证当用户向服务器发起请求时,会对token是否有效进行验证
微服务事务理论分布式系统无法完成三个指标,即CAP定理 Consistency(一致性):各服务之间数据一致 Availability(可用性):访问服务必须得到响应 Partition Tolerance(分区容错性):各服务之间异步执行,互不影响 事务问题在微服务中,一个服务出现问题,其关联的服务仍然会继续运行,这就容易出现事务问题出现这种原因是分布式不可避免的CAP定理,即分布式服务中(P),一旦A服务宕机,B服务为保持可用性(A),不会也跟着宕机,那么数据(C)就不一致,产生事务问题解决方法 AP模式:各服务在运行过程中,一旦出错则其关联服务执行弥补措施,即同时保证可用性(A)和分布式(P) 例:创建订单和扣减余额正常运行,但是扣减库存出现问题,那么就删除订单并添加余额 CP模式:各服务在运行过程中,等待其关联服务运行完成后再一起提交,即同时保证一致性(C)和分布式(P) 例:创建订单和扣减余额运行完成后等待扣减库存,扣减库存出现问题,则不提交事务无论是AP还是CP,都需要一个中间商检测各服务的健康状况,从而判断执行什么操作,因此我们需要一个事务协调者,即Seata认识Seata介绍Seata是一个微服务/分布式的事务管理框架,其主要有3各部分组成: TC-事务协调者:维护全局事务状态,协调全局事务提交或回滚 TM-事务管理器:定义全局事务范围、定义全局事务、提交或回滚全局事务 RM-资源管理器:管理分支事务资源,与TC交谈以注册事务和报告事务状态,并驱动分支事务提交或回滚 部署TC服务1.下载TC服务首先我们要下载seata-server包，地址在http://seata.io/zh-cn/blog/download.html2.修改本地配置```properties registry { # tc服务的注册中心类，这里选择nacos，也可以是eureka、zookeeper等 type = "nacos"nacos { # seata tc 服务注册到 nacos的服务名称，可以自定义 application = "seata-tc-server" serverAddr = "127.0.0.1:8848" group = "DEFAULT_GROUP" namespace = "" cluster = "SH" username = "nacos" password = "nacos" } }config { # 读取tc服务端的配置文件的方式，这里是从nacos配置中心读取，这样如果tc是集群，可以共享配置 type = "nacos" # 配置nacos地址等信息 nacos { serverAddr = "127.0.0.1:8848" namespace = "" group = "SEATA_GROUP" username = "nacos" password = "nacos" dataId = "seataServer.properties" } } ```3.nacos添加配置```properties数据存储方式，db代表数据库store.mode=db store.db.datasource=druid store.db.dbType=mysql store.db.driverClassName=com.mysql.jdbc.Driver store.db.url=jdbc:mysql://127.0.0.1:3306/seata?useUnicode=true&rewriteBatchedStatements=true store.db.user=root store.db.password=123 store.db.minConn=5 store.db.maxConn=30 store.db.globalTable=global_table store.db.branchTable=branch_table store.db.queryLimit=100 store.db.lockTable=lock_table store.db.maxWait=5000事务、日志等配置server.recovery.committingRetryPeriod=1000 server.recovery.asynCommittingRetryPeriod=1000 server.recovery.rollbackingRetryPeriod=1000 server.recovery.timeoutRetryPeriod=1000 server.maxCommitRetryTimeout=-1 server.maxRollbackRetryTimeout=-1 server.rollbackRetryTimeoutUnlockEnable=false server.undo.logSaveDays=7 server.undo.logDeletePeriod=86400000客户端与服务端传输方式transport.serialization=seata transport.compressor=none关闭metrics功能，提高性能metrics.enabled=false metrics.registryType=compact metrics.exporterList=prometheus metrics.exporterPrometheusPort=9898 ```4.创建数据库tc服务在管理分布式事务时，需要记录事务相关数据到数据库中，需要提前创建好这些表。这些表主要记录全局事务、分支事务、全局锁信息：```mysqlSET NAMES utf8mb4; SET FOREIGN_KEY_CHECKS = 0;-- 分支事务表DROP TABLE IF EXISTS branch_table; CREATE TABLE branch_table ( branch_id bigint(20) NOT NULL, xid varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL, transaction_id bigint(20) NULL DEFAULT NULL, resource_group_id varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, resource_id varchar(256) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, branch_type varchar(8) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, status tinyint(4) NULL DEFAULT NULL, client_id varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, application_data varchar(2000) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, gmt_create datetime(6) NULL DEFAULT NULL, gmt_modified datetime(6) NULL DEFAULT NULL, PRIMARY KEY (branch_id) USING BTREE, INDEX idx_xid(xid) USING BTREE ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;-- 全局事务表DROP TABLE IF EXISTS global_table; CREATE TABLE global_table ( xid varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL, transaction_id bigint(20) NULL DEFAULT NULL, status tinyint(4) NOT NULL, application_id varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, transaction_service_group varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, transaction_name varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, timeout int(11) NULL DEFAULT NULL, begin_time bigint(20) NULL DEFAULT NULL, application_data varchar(2000) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, gmt_create datetime NULL DEFAULT NULL, gmt_modified datetime NULL DEFAULT NULL, PRIMARY KEY (xid) USING BTREE, INDEX idx_gmt_modified_status(gmt_modified, status) USING BTREE, INDEX idx_transaction_id(transaction_id) USING BTREE ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;-- 全局锁DROP TABLE IF EXISTS lock_table; CREATE TABLE lock_table ( row_key VARCHAR(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL, xid VARCHAR(96) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, transaction_id BIGINT(20) NULL DEFAULT NULL, branch_id BIGINT(20) NOT NULL, resource_id VARCHAR(256) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, table_name VARCHAR(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, pk VARCHAR(36) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, gmt_create DATETIME NULL DEFAULT NULL, gmt_modified DATETIME NULL DEFAULT NULL, PRIMARY KEY (row_key) USING BTREE, INDEX idx_branch_id(branch_id) USING BTREE ) ENGINE = INNODB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = COMPACT;SET FOREIGN_KEY_CHECKS = 1; ```5.启动TC服务进入bin目录，运行其中的seata-server.bat即可：微服务集成Seata(部署TM,RM)1.引入依赖```xml com.alibaba.cloud spring-cloud-starter-alibaba-seata seata-spring-boot-starter io.seata io.seata seata-spring-boot-starter ${seata.version} ```2.修改配置文件yaml seata: registry: # TC服务注册中心的配置，微服务根据这些信息去注册中心获取tc服务地址 # 参考tc服务自己的registry.conf中的配置 type: nacos nacos: # tc server-addr: 127.0.0.1:8848 namespace: "" group: DEFAULT_GROUP application: seata-tc-server # tc服务在nacos中的服务名称 cluster: SH tx-service-group: seata-demo # 事务组，根据这个获取tc服务的cluster名称 service: vgroup-mapping: # 事务组与TC服务cluster的映射关系 seata-demo: SH解决方法XA模式XA模式是一种CP模式(强一致性),等待所有服务执行完成后再一并回滚seata对XA进行了一点小调整,但大体相同yaml srata: data-source-proxy-mode: XAjade @service public class xxServiceImpl implements xxService{ @override @GlobalTransactional public void xxMethod(){ ... } }AT模式(默认)AT模式是XA模式的改进方案,属于AP模式(强可用性),不等待其他服务,先记录原始数据,再直接提交,若失败则修改数据AT使用一个全局锁,保证在同一时间修改数据时只能由一个事务处理(下图假设事务1第2此提交失败)由一个极端情况,就是同一时间两个修改数据时,另一个事务不受seata管理,这时seata会报错,业务上我们捕捉这个异常,然后人工处理(途中事务1第2次提交失败)在使用前需要先在微服务数据库里创建快照undolog(不是seata数据库)``sql DROP TABLE IF EXISTSundo_log; CREATE TABLEundo_log(branch_idbigint(20) NOT NULL COMMENT 'branch transaction id',xidvarchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT 'global transaction id',contextvarchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT 'undo_log context,such as serialization',rollback_infolongblob NOT NULL COMMENT 'rollback info',log_statusint(11) NOT NULL COMMENT '0:normal status,1:defense status',log_createddatetime(6) NOT NULL COMMENT 'create datetime',log_modifieddatetime(6) NOT NULL COMMENT 'modify datetime', UNIQUE INDEXux_undo_log(xid,branch_id`) USING BTREE ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = 'AT transaction mode undo table' ROW_FORMAT = Compact;```yaml srata: data-source-proxy-mode: ATjava @service public class xxServiceImpl implements xxService{ @override @GlobalTransactional public void xxMethod(){ ... } }TCC模式TCC模式是AT模式的手动挡,AT模式会提前保存原始数据,错误后恢复原始数据.而TCC模式则是分为三个模式,Try,Confirm,Cancel Try:事务提交 Confirm:提交成功后执行 Cancel:提交失败后执行 且TCC没有用锁,性能是最好的实现空回滚从而防止业务悬挂,就是各模式没有走正常流程(Try->Confirm/Cancel)而直接执行,我们需要自定义一张表来记录数据状态.在执行Confirm/Cancel前先判断状态状态表需要自己定义,整体结构可参考此表sql DROP TABLE IF EXISTS `account_freeze_tbl`; CREATE TABLE `account_freeze_tbl` ( `xid` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL, `user_id` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, `freeze_money` int(11) UNSIGNED NULL DEFAULT 0, `state` int(1) NULL DEFAULT NULL COMMENT '事务状态，0:try，1:confirm，2:cancel', PRIMARY KEY (`xid`) USING BTREE ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = COMPACT;整体设计流程如图:```java @LocalTCC public interface xxxService{ //Try @TwoPhaseBusinessAction(name="try方法名称",commitMethod ="confirm方法名称" ,rollbackMethod ="cancel方法名称" ) void Try方法(@BusinessActionContextParameter 参数); //在需要上下文获取的数据使用@BusinessActionContextParameter //confirm和cancel想要拿到参数,就需要给参数@B...Contex//context可以获取事务信息和参数 Bollean confirm方法(BusinessActionContext context); Bollean cancel方法(BusinessActionContext context); } ```java @service public class xxServiceImpl implements xxService{ @override @Transactional public void Try方法(){ // 1. 获取事务信息 int xid = RootContext.getXID(); // 2. 查看状态表中是否已经有事务,有则返回空(防止事务悬挂) ... // 3. 执行事务 ... // 4. 向状态表中添加信息 ... } @override public void confirm方法(BusinessActionContext context){ // 1. 获取事务信息 int xid = context.getXid; // 2. 判断状态表,查看是否存在且状态为Try ... // 3. 删除状态 ... } @override public void cancel方法(BusinessActionContext context){ // 1. 获取事务信息 int xid = context.getXid; // 2. 判断状态表,查看是否存在且状态为Try ... //2.2若不存在,则空回滚,向状态表中添加空信息 //context.getActionContext("参数")获取参数 // 3. 恢复数据,状态改为Cancel ... } }SAGA模式该模式由于没有事务隔离,非常危险,不建议使用