1、jvm 内存模型程序计数器 Program Counter Register：记录的是正在执行的虚拟机字节码指令的地址。此内存区域是唯一一个在JAVA虚拟机规范中没有规定任何OutOfMemoryError的区域虚拟机：VM Stack描述的是 JAVA 方法执行的内存模型，每个方法在执行的时候都会创建一个栈帧， 用于存储局部变量表，操作数栈，动态链接，方法接口等信息局部变量表存储了编译期可知的各种基本数据类型、对象引用线程请求的栈深度不够会报 StackOverflowError 异常栈动态扩展的容量不够会报 OutOfMemoryError 异常虚拟机栈是线程隔离的，即每个线程都有自己独立的虚拟机栈本地方法：Native Stack本地方法栈类似于虚拟机栈，只不过本地方法栈使用的是本地方法堆：Heap几乎所有的对象实例都在堆上分配内存2、堆所有的对象实例以及数组都要在堆上分配。堆是垃圾收集器管理的主要区域，也被称为“GC 堆”；也是我们优化最多考虑的地方。 堆可以细分为：1.新生代Eden 空间From Survivor 空间To Survivor 空间2.老年代3.永久代/元空间Java8 以前永久代，受 jvm 管理，java8 以后元空间，直接使用物理内存。因此， 默认情况下，元空间的大小仅受本地内存限制。 垃圾回收从 Java8 开始，HotSpot 已经完全将永久代（Permanent Generation）移除，取而代之的是一 个新的区域—元空间（MetaSpace）3、运行图由于FGC的速度比YGC的速度慢得多，所以JVM的优化也是尽量避免FGC
1.给es中建立索引。product,建立好映射关系PUT product { "mappings": { "properties": { "skuId": { "type": "long" }, "spuId": { "type": "long" }, "skuTitle": { "type": "text", "analyzer": "ik_smart" }, "skuPrice": { "type": "keyword" }, "skuImg": { "type": "keyword", "index": false, "doc_values": false }, "saleCount": { "type": "long" }, "hosStock": { "type": "boolean" }, "hotScore": { "type": "long" }, "brandId": { "type": "long" }, "catelogId": { "type": "long" }, "brandName": { "type": "keyword", "index": false, "doc_values": false }, "brandImg": { "type": "keyword", "index": false, "doc_values": false }, "catelogName": { "type": "keyword", "index": false, "doc_values": false }, "attrs": { "type": "nested", "properties": { "attrId": { "type": "long" }, "attrName": { "type": "keyword", "index": false, "doc_values": false }, "attrValue": { "type": "keyword" } } } } } }2.Service类package com.atguigu.gulimall.lsearch.service; import com.atguigu.common.to.es.SkuEsModel; import java.io.IOException; import java.util.List; public interface ProductSaveService { boolean productStatusUp(List<SkuEsModel> skuEsModels) throws IOException; }3.ServiceImpl类package com.atguigu.gulimall.lsearch.service.iml; import com.alibaba.fastjson.JSON; import com.atguigu.common.to.es.SkuEsModel; import com.atguigu.gulimall.lsearch.config.GulimallElasticSearchConfig; import com.atguigu.gulimall.lsearch.constant.EsConstant; import com.atguigu.gulimall.lsearch.service.ProductSaveService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.bulk.BulkRequest; import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.xcontent.XContentType; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.Arrays; import java.util.List; import java.util.stream.Collectors; @Slf4j @Service("productSaveService") public class ProductSaveServiceImpl implements ProductSaveService { @Autowired RestHighLevelClient restHighLevelClient; @Override public boolean productStatusUp(List<SkuEsModel> skuEsModels) throws IOException { //保存到es //1.给es中建立索引。product,建立好映射关系 //2.给es中保存数据 BulkRequest bulkRequest, RequestOptions options BulkRequest bulkRequest = new BulkRequest(); for (SkuEsModel model : skuEsModels) { //构造保存请求 IndexRequest indexRequest = new IndexRequest(EsConstant.PRODUCT_INDEX); indexRequest.id(model.getSkuId().toString()); String s = JSON.toJSONString(model); indexRequest.source(s, XContentType.JSON); bulkRequest.add(indexRequest); } BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, GulimallElasticSearchConfig.COMMON_OPTIONS); //TODO 如果批量保存错误则需要处理 boolean b = bulk.hasFailures(); List<String> collect = Arrays.stream(bulk.getItems()).map(item -> { return item.getId(); }).collect(Collectors.toList()); log.info("商品上架成功:{}，返回数据:{}", collect,bulk.toString()); return b; } }4.Controller类package com.atguigu.gulimall.lsearch.controller; import com.atguigu.common.exception.BizCodeEnume; import com.atguigu.common.to.es.SkuEsModel; import com.atguigu.common.utils.R; import com.atguigu.gulimall.lsearch.service.ProductSaveService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import java.util.List; @Slf4j @RequestMapping("/search/save") @RestController public class ElasticSaveController { @Autowired ProductSaveService productSaveService; @PostMapping("/product") public R productStatusUp(@RequestBody List<SkuEsModel> skuEsModels) { boolean b = false; try { b = productSaveService.productStatusUp(skuEsModels); } catch (Exception e) { log.error("ElasticSaveController商品商家错误:{}", e); return R.error(BizCodeEnume.PRODUCT_UP_EXCEPTION.getCode(), BizCodeEnume.PRODUCT_UP_EXCEPTION.getMessage()); } if (!b) { return R.ok(); } else { return R.error(BizCodeEnume.PRODUCT_UP_EXCEPTION.getCode(), BizCodeEnume.PRODUCT_UP_EXCEPTION.getMessage()); } } }5.BizCodeEnume枚举package com.atguigu.common.exception; /** * @Description: 错误状态码枚举 * @Created: with IntelliJ IDEA. * @author: 夏沫止水 * @createTime: 2020-05-27 17:29 * * 错误码和错误信息定义类 * 1. 错误码定义规则为5为数字 * 2. 前两位表示业务场景，最后三位表示错误码。例如：100001。10:通用 001:系统未知异常 * 3. 维护错误码后需要维护错误描述，将他们定义为枚举形式 * 错误码列表： * 10: 通用 * 001：参数格式校验 * 002：短信验证码频率太高 * 11: 商品 * 12: 订单 * 13: 购物车 * 14: 物流 * 15：用户 **/ public enum BizCodeEnume { UNKNOW_EXCEPTION(10000, "系统未知异常"), VAILD_EXCEPTION(10001, "参数格式校验失败"), TO_MANY_REQUEST(10002, "请求流量过大，请稍后再试"), SMS_CODE_EXCEPTION(10002, "验证码获取频率太高，请稍后再试"), PRODUCT_UP_EXCEPTION(11000, "商品上架异常"), USER_EXIST_EXCEPTION(15001, "存在相同的用户"), PHONE_EXIST_EXCEPTION(15002, "存在相同的手机号"), NO_STOCK_EXCEPTION(21000, "商品库存不足"), LOGINACCT_PASSWORD_EXCEPTION(15003, "账号或密码错误"), ; private Integer code; private String message; BizCodeEnume(Integer code, String message) { this.code = code; this.message = message; } public Integer getCode() { return code; } public String getMessage() { return message; } }6.其他测试功能package com.atguigu.gulimall.lsearch; import com.alibaba.fastjson.JSON; import com.atguigu.gulimall.lsearch.config.GulimallElasticSearchConfig; import lombok.Data; import lombok.ToString; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.aggregations.Aggregation; import org.elasticsearch.search.aggregations.AggregationBuilders; import org.elasticsearch.search.aggregations.Aggregations; import org.elasticsearch.search.aggregations.bucket.terms.Terms; import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder; import org.elasticsearch.search.aggregations.metrics.Avg; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; import java.io.IOException; import java.util.Map; @SpringBootTest public class GulimallSearchApplicationTests { @Autowired private RestHighLevelClient client; /** * 测试复杂查询功能 */ @Test public void searchDate() throws IOException { //1.创建检索请求 SearchRequest searchRequest = new SearchRequest(); //指定索引 searchRequest.indices("bank"); //指定DSL，检索条件 //SearchSourceBuilder sourceBuilder 封装的条件 SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); //1.1构造检索条件 searchSourceBuilder.query(); searchSourceBuilder.from(); searchSourceBuilder.size(); searchSourceBuilder.aggregation(); searchSourceBuilder.query(QueryBuilders.matchQuery("address", "mill")); //1.2按照年龄的值分布进行聚合 TermsAggregationBuilder ageAgg = AggregationBuilders.terms("ageAgg").field("age").size(10); searchSourceBuilder.aggregation(ageAgg); //1.3计算平均薪资 TermsAggregationBuilder balanceAvg = AggregationBuilders.terms("balanceAvg").field("balance"); searchSourceBuilder.aggregation(balanceAvg); System.out.println("检索条件" + searchSourceBuilder.toString()); searchRequest.source(searchSourceBuilder); //2.执行检索 SearchResponse searchResponce = client.search(searchRequest, GulimallElasticSearchConfig.COMMON_OPTIONS); //3.分析结果 System.out.println(searchResponce.toString()); //Map map = JSON.parseObject(searchResponce.toString(), Map.class); 这种方式太麻烦 //3.1获取所有查询到的数据 SearchHits hits = searchResponce.getHits(); SearchHit[] searchHits = hits.getHits(); for (SearchHit hit : searchHits) { String sourceAsString = hit.getSourceAsString(); Account account = JSON.parseObject(sourceAsString, Account.class); System.out.println("account" + account); } //3.2获取这次检索到的分析信息 Aggregations aggregations = searchResponce.getAggregations(); Terms ageAgg1 = aggregations.get("ageAgg"); for (Terms.Bucket bucket : ageAgg1.getBuckets()) { String keyAsString = bucket.getKeyAsString(); System.out.println("年龄" + keyAsString + "===>" + bucket.getDocCount()); } Avg balanceAvg1 = aggregations.get("balanceAvg"); System.out.println("平均工资" + balanceAvg1.getValue()); } @ToString @Data static class Account { private int account_number; private int balance; private String firstname; private String lastname; private int age; private String gender; private String address; private String employer; private String email; private String city; private String state; } /** * 测试存储数据到es,保存更新二合一 */ @Test public void indexDate() throws IOException { IndexRequest indexRequest = new IndexRequest("users"); //数据的id indexRequest.id("1"); // indexRequest.source("userName", "youzi", "age", 18); User user = new User(); user.setUserName("Youzi"); user.setAge(28); user.setGender("男"); String jsonString = JSON.toJSONString(user); //要保存的内容 indexRequest.source(jsonString, XContentType.JSON); //执行index操作 IndexResponse index = client.index(indexRequest, GulimallElasticSearchConfig.COMMON_OPTIONS); //提取有用的响应数据 System.out.println(index); } @Data class User { private String userName; private String gender; private int age; } }7.工具类package com.atguigu.gulimall.lsearch.config; import org.apache.http.HttpHost; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestHighLevelClient; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * 1.导入配置 * 2.编写配置,给容器内注入一个RestHighLevelClient */ @Configuration public class GulimallElasticSearchConfig { @Bean public RestHighLevelClient esRestClient() { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( //第一个参数是ElasticSearch安装机地址，第二个参数是端口，第三个参数是协议"http"或"https" new HttpHost("192.168.56.10", 9200, "http") ) ); return client; } public static final RequestOptions COMMON_OPTIONS; static { RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder(); // builder.addHeader("Authorization", "Bearer " + TOKEN); // builder.setHttpAsyncResponseConsumerFactory( // new HttpAsyncResponseConsumerFactory // .HeapBufferedResponseConsumerFactory(30 * 1024 * 1024 * 1024)); COMMON_OPTIONS = builder.build(); } }
继承Thread类创建线程//总结：重写的是run()方法，而不是start()方法，但是占用了继承的名额，Java中的类是单继承的 public class MyThread extends Thread{ public static void main(String[] args){ MyThread myThread = new MyThread(); //创建并启动线程 myThread.start(); } @Override public void run(){ //重写run方法 System.out.println("Hello"); } }实现Runnable接口创建线程//总结：实现Runnable接口，实现run()方法，使用依然要用到Thread类，这种方式更常用 //第一种写法 public class MyThread implements Runnable { public static void main(String[] args){ Thread thread = new Thread(new MyThread()); //创建并启动线程 thread.start(); } @Override public void run(){ //重写run方法 System.out.println("Hello"); } } //第二种写法 public class MyThread{ public static void main(String[] args){ Thread thread = new Thread(new Runnable(){ public void run(){ //重写run方法 System.out.println("Hello"); } }); //创建并启动线程 thread.start(); } } //第三种写法 public class MyThread implements Runnable { public static void main(String[] args){ Thread thread = new Thread(()->System.out.println("Hello")); thread.start(); } }使用Callable和Future创建线程//总结：与Runnable的区别在于，Callable可以获取返回结果 //实现Callable接口，实现call()方法，要使用Thread和FutureTask配合，这种方式支持拿到异步执行结果 public class MyThread implements Callable<String> { public static void main(String[] args) throws ExecutionException, InterruptedException { FutureTask<String> futureTask = new FutureTask<>(new MyThread()); Thread thread = new Thread(futureTask); thread.start(); //此时是获取到的结果 String result = futureTask.get(); System.out.println(result); } @Override public String call() throws Exception { return "Hello"; } }使用线程池例如用Executor框架//总结：实现Callable接口或者实现Runnable接口都可以，由ExecutorService创建线程 public class MyThread implements Runnable { public static void main(String[] args){ // 创建10个数据级的线程池 ExecutorService threadPool = Executors.newFixedThreadPool(10); threadPool.execute(new MyThread()); } @Override public void run(){ System.out.println("Hello"); } }
解决POST、GET请求中文乱码问题解决post请求乱码问题： 在web.xml中配置一个CharacterEncodingFilter过滤器，设置成 utf-8；<filter> <filter-name>CharacterEncodingFilter</filter-name> <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class> <init-param> <param-name>encoding</param-name> <param-value>utf-8</param-value> </init-param> </filter> <filter-mapping> <filter-name>CharacterEncodingFilter</filter-name> <url-pattern>/*</url-pattern> </filter-mapping>@RequestMapping(value = "/add", produces = "text/plain;charset=utf-8")get 请求中文参数出现乱码解决方法有两个修改 tomcat 配置文件添加编码与工程编码一致，如下：<ConnectorURIEncoding="utf-8" connectionTimeout="20000" port="8080" protocol="HTTP/1.1" redirectPort="8443"/另外一种方法对参数进行重新编码//ISO8859-1 是 tomcat 默认编码，需要将 tomcat 编码后的内容按 utf-8 编码 String userName = new String(request.getParamter(“userName”).getBytes(“ISO8859-1”),“utf-8”)RedisRedis作用内存存储、持久化，放在内存中，断电即失（持久化重要（RDB、AOF））效率高，可以用于告诉缓存发布订阅系统地图信息分析计数器、计数器（浏览量）特性多样的数据类型持久化集群事务基础知识redis默认16个数据库，默认使用第0个，可以使用select进行切换数据库常用命令#查看数据库大小 DBSIZE #查看所有的key KEY * #清空当前数据库 flushdb #清空所有数据库 flushallRedis是单线程的Redis是基于内存操作，所以他的瓶颈在于机器的内存和网络带宽SpringBean的配置详解Bean的实例化配置静态工厂方法实例化Bean实例工厂方法实例化Bean实现FactoryBean规范延迟实例化BeanBean的依赖注入配置依赖注入的两种方式通过Bean的set方法注入<property name="" ref="" /><property name="" value="" />通过构造Bean的方法注入<constructor-arg name="" ref="" /><constructor-arg name="" value="" />ref用于引用其他的Bean的id；value用于注入普通属性值.依赖注入的数据类型有如下三种：普通数据类型，例如：String，int，boolean等，通过value属性指定引用数据类型，例如：UserDaoImpl，DataSource等，通过ref属性指定集合数据类型，例如：List，Map，Properties等Spring 声明式事务无效可能的原因有哪些？答：可能的原因如下：MySQL 使用的是 MyISAM 引擎，而 MyISAM 是不支持事务的；@Transactional 使用在非 public 方法上，@Transactional 注解只能支持 public 级别，其他类型声明的事务不会生效；@Transactional 在同一个类中无事务方法 A() 内部调用有事务方法 B()，那么此时 B() 事物不会生效。Spring 中的 AOP 的底层实现原理是什么？答：Spring AOP 的底层实现原理就是动态代理。Spring AOP 的动态代理有两种实现方式，对于接口使用的是 JDK 自带的动态代理来实现的，而对比非接口使用的是 CGLib 来实现的。Spring 中 Bean 的生命周期答：Spring 中 Bean 的生命周期如下：① 实例化 Bean：对于 BeanFactory 容器，当客户向容器请求一个尚未初始化的 Bean 时，或初始化 Bean 的时候需要注入另一个尚未初始化的依赖时，容器就会调用 createBean 进行实例化。对于 ApplicationContext 容器，当容器启动结束后，通过获取 BeanDefinition 对象中的信息，实例化所有的 Bean；② 设置对象属性（依赖注入）：实例化后的对象被封装在 BeanWrapper 对象中，紧接着 Spring 根据 BeanDefinition 中的信息以及通过 BeanWrapper 提供的设置属性的接口完成依赖注入；③ 处理 Aware 接口：Spring 会检测该对象是否实现了 xxxAware 接口，并将相关的 xxxAware 实例注入给 Bean：如果这个 Bean 已经实现了 BeanNameAware 接口，会调用它实现的 setBeanName(String BeanId) 方法，此处传递的就是 Spring 配置文件中 Bean 的 id 值；如果这个 Bean 已经实现了 BeanFactoryAware 接口，会调用它实现的 setBeanFactory() 方法，传递的是 Spring 工厂自身；如果这个 Bean 已经实现了 ApplicationContextAware 接口，会调用 setApplicationContext(ApplicationContext) 方法，传入 Spring 上下文；④ BeanPostProcessor：如果想对 Bean 进行一些自定义的处理，那么可以让 Bean 实现了 BeanPostProcessor 接口，那将会调用 postProcessBeforeInitialization(Object obj, String s) 方法；⑤ InitializingBean 与 init-method：如果 Bean 在 Spring 配置文件中配置了 init-method 属性，则会自动调用其配置的初始化方法；⑥ 如果这个 Bean 实现了 BeanPostProcessor 接口，将会调用 postProcessAfterInitialization(Object obj, String s) 方法；由于这个方法是在 Bean 初始化结束时调用的，因而可以被应用于内存或缓存技术；以上几个步骤完成后，Bean 就已经被正确创建了，之后就可以使用这个 Bean 了。⑦ DisposableBean：当 Bean 不再需要时，会经过清理阶段，如果 Bean 实现了 DisposableBean 这个接口，会调用其实现的 destroy() 方法；⑧ destroy-method：最后，如果这个 Bean 的 Spring 配置中配置了 destroy-method 属性，会自动调用其配置的销毁方法。Spring 中都是用到的设计模式答：Spring 中使用的设计模式如下：工厂模式：通过 BeanFactory、ApplicationContext 来创建 bean 都是属于工厂模式；单例、原型模式：创建 bean 对象设置作用域时，就可以声明 Singleton（单例模式）、Prototype（原型模式）；察者模式：Spring 可以定义一下监听，如 ApplicationListener 当某个动作触发时就会发出通知；责任链模式：AOP 拦截器的执行；策略模式：在创建代理类时，如果代理的是接口使用的是 JDK 自身的动态代理，如果不是接口使用的是 CGLIB 实现动态代理。设置 Ehcache 为 MyBatis 的二级缓存直接在 XML 中配置开启 EhcacheCache<mapper namespace="com.interview.repository.ClassesReposirory"> <cache type="org.mybatis.caches.ehcache.EhcacheCache" > <property name="timeToIdleSeconds" value="3600"/> <property name="timeToLiveSeconds" value="3600"/> <property name="memoryStoreEvictionPolicy" value="LRU"/> </cache> select * from classes where id = #{id} </mapper>反射的基本使用下来我们通过反射调用类中的某个方法，来学习反射的基本使用。使用反射调用类中的方法，分为三种情况：调用静态方法调用公共方法调用私有方法假设有一个实体类 MyReflect 包含了以上三种方法，代码如下：package com.interview.chapter4; class MyReflect { // 静态方法 public static void staticMd() { System.out.println("Static Method"); } // 公共方法 public void publicMd() { System.out.println("Public Method"); } // 私有方法 private void privateMd() { System.out.println("Private Method"); } }下面分别来看，使用反射如何调用以上三种类型的方法。① 反射调用静态方法Class myClass = Class.forName("com.interview.chapter4.MyReflect"); Method method = myClass.getMethod("staticMd"); method.invoke(myClass);② 反射调用公共方法Class myClass = Class.forName("com.interview.chapter4.MyReflect"); // 创建实例对象（相当于 new ） Object instance = myClass.newInstance(); Method method2 = myClass.getMethod("publicMd"); method2.invoke(instance);③ 反射调用私有方法Class myClass = Class.forName("com.interview.chapter4.MyReflect"); // 创建实例对象（相当于 new ） Object object = myClass.newInstance(); Method method3 = myClass.getDeclaredMethod("privateMd"); method3.setAccessible(true); method3.invoke(object);反射使用总结反射获取调用类可以通过 Class.forName()，反射获取类实例要通过 newInstance()，相当于 new 一个新对象，反射获取方法要通过 getMethod()，获取到类方法之后使用 invoke() 对类方法进行调用。如果是类方法为私有方法的话，则需要通过 setAccessible(true) 来修改方法的访问限制，以上的这些操作就是反射的基本使用。
监听器（用MyBatis的IService接口做了封装）package io.github.talelin.latticy.listener; import com.alibaba.excel.context.AnalysisContext; import com.alibaba.excel.event.AnalysisEventListener; import com.alibaba.excel.util.ListUtils; import com.alibaba.fastjson.JSON; import com.baomidou.mybatisplus.extension.service.IService; import lombok.extern.slf4j.Slf4j; import java.util.List; import java.util.Map; @Slf4j public class ExcelReaderListener<T> extends AnalysisEventListener<T> { /** * 每隔5条存储数据库，实际使用中可以100条，然后清理list ，方便内存回收 */ private static final int BATCH_COUNT = 100; /** * 缓存的数据 */ private List<T> cachedDataList = ListUtils.newArrayListWithExpectedSize(BATCH_COUNT); /** * 假设这个是一个DAO，当然有业务逻辑这个也可以是一个service。当然如果不用存储这个对象没用。 */ private IService<T> service; /** * 如果使用了spring,请使用这个构造方法。每次创建Listener的时候需要把spring管理的类传进来 * * @param service */ public ExcelReaderListener(IService<T> service) { this.service = service; } /** * 这个每一条数据解析都会来调用 * * @param data one row value. Is is same as {@link AnalysisContext#readRowHolder()} * @param context */ @Override public void invoke(T data, AnalysisContext context) { log.info("解析到一条数据:{}", JSON.toJSONString(data)); cachedDataList.add(data); // 达到BATCH_COUNT了，需要去存储一次数据库，防止数据几万条数据在内存，容易OOM if (cachedDataList.size() >= BATCH_COUNT) { saveData(); // 存储完成清理 list //注意ListUtils.newArrayListWithExpectedSize是在EasyExcel3.1.1版本之后才有的 cachedDataList = ListUtils.newArrayListWithExpectedSize(BATCH_COUNT); } } @Override public void invokeHeadMap(Map<Integer, String> headMap, AnalysisContext context) { log.info("表头数据 excelHead= {}", headMap); } /** * 所有数据解析完成了 都会来调用 * * @param context */ @Override public void doAfterAllAnalysed(AnalysisContext context) { // 这里也要保存数据，确保最后遗留的数据也存储到数据库 saveData(); log.info("所有数据解析完成！"); } /** * 加上存储数据库 */ private void saveData() { log.info("{}条数据，开始存储数据库！", cachedDataList.size()); service.saveBatch(cachedDataList); log.info("存储数据库成功！"); } }控制层代码@PostMapping("/import") public AjaxResult importExcel(MultipartFile file) throws IOException { AjaxResult ajaxResult = new AjaxResult(); try { //file.getInputStream()是获取导入的文件数据 //SStbkSmbsnFncDtlI01ADO.class是对应的实体类 //YwtzTyrzmxCwbListener是对应的监听器，这里是自定义对导入的数据做判断限制 //sStbkSmbsnFncDtlI01AService是对应的Service类 //.sheet()第一个参数是Excel哪个中sheet，第二个参数是sheet名称 //.headRowNumber()是表头行数 //.doRead()开始读取 EasyExcel.read(file.getInputStream(), SStbkSmbsnFncDtlI01ADO.class, new YwtzTyrzmxCwbListener(sStbkSmbsnFncDtlI01AService)).sheet(1, "02表内容").headRowNumber(3).doRead(); return ajaxResult.success("success"); } catch (ExcelAnalysisException e) { e.printStackTrace(); return ajaxResult.fail("500", e.getMessage()); } }实体类package io.github.talelin.latticy.model; import com.alibaba.excel.annotation.ExcelIgnore; import com.alibaba.excel.annotation.ExcelProperty; import com.baomidou.mybatisplus.annotation.TableName; import lombok.Data; import lombok.EqualsAndHashCode; import java.io.Serializable; import java.math.BigDecimal; /** * B_H01_同业融资信息_财务部_补录 * * @author generator@TaleLin * @since 2022-09-05 */ @Data public class SStbkSmbsnFncInfH01ADO implements Serializable { /** * 客户号 */ //强制读取第一个 //这里不建议 index 和 name 同时用，要么一个对象只用index，要么一个对象只用name去匹配 //用名字去匹配，这里需要注意，如果名字重复，会导致只有一个字段读取到数据 @ExcelProperty(index = 0) private String cstNo; /** * 业务编码 */ @ExcelProperty(index = 1) private String bsnEcd; /** * 业务种类 */ @ExcelProperty(index = 2) private String bsnCtlg; /** * 起始日期 */ @ExcelProperty(index = 3) private String begDt; /** * 到期日 */ @ExcelProperty(index = 4) private String exdy; /** * 实际终止日期 */ @ExcelProperty(index = 5) private String actTmdt; /** * 期限 */ @ExcelProperty(index = 6) private String ddln; /** * 利率类型 */ @ExcelProperty(index = 7) private String intrtTp; /** * 实际利率 */ @ExcelProperty(index = 8) private BigDecimal actIntrt; /** * 借贷定价基准类型 */ @ExcelProperty(index = 9) private String dbtcrPrcgBssTp; /** * 基准利率 */ @ExcelProperty(index = 10) private BigDecimal bssIntrt; /** * 计息方式 */ @ExcelProperty(index = 11) private String intarMod; /** * 利率浮动频率 */ @ExcelProperty(index = 12) private String intrtFltFrq; /** * 余额 */ @ExcelProperty(index = 13) private BigDecimal bal; /** * 内部机构号 */ @ExcelProperty(index = 14) private String inrInsno; /** * 币种 */ @ExcelProperty(index = 15) private String ccy; /** * 数据来源 */ @ExcelIgnore private String dtsrc; /** * 业务日期 */ //忽略这个字段 @ExcelIgnore private String parDate; /** * 源系统代码 */ @ExcelIgnore private String srcSysCd; /** * 加载时间戳 */ @ExcelIgnore private String loadTime; /** * 金融机构代码 */ @ExcelProperty(index = 16) private String fncLcnsNo; /** * 交易对手代码 */ @ExcelProperty(index = 17) private String txnOpntCd; /** * 交易对手证件类别 */ @ExcelProperty(index = 18) private String txnOpntCrdtCgy; /** * 资产负债类型 */ @ExcelProperty(index = 19) private String alTp; /** * 余额折人民币 */ @ExcelProperty(index = 20) private BigDecimal balFoldCny; /** * 存款协议代码 */ @ExcelProperty(index = 21) private String depAgrmCd; /** * 缴存准备金方式 */ @ExcelProperty(index = 22) private String depRsrvfndMtd; /** * 合同编码 */ @ExcelProperty(index = 23) private String ctrEcd; }
在AOP配置事务控制或注解 @Transactional控制事务中，try…catch…使事务失效的解决办法
可在catch中抛出运行时异常throw new RuntimeException(e)（注：默认回滚的是RuntimeException，如果你想触发其他异常的回滚，需要在注解上配置一下）手动回滚TransactionAspectSupport.currentTransactionStatus().setRollbackOnly(),使得事务生效，出现异常回滚。@Transactional(rollbackFor = Exception.class) public AjaxResult update(@RequestBody RestaurantVo entity) { AjaxResult ajaxResult = new AjaxResult(); try { service.save(entity); return ajaxResult.success("success"); } catch (Exception e) { //throw new RuntimeException(e); TransactionAspectSupport.currentTransactionStatus().setRollbackOnly(); return ajaxResult.fail("500", e.getMessage()); } }
1、什么是字节流，什么是字符流字节流： 它处理单元为1个字节（byte），操作字节和字节数组，存储的是二进制文件，如果是音频文件、图片、歌曲，就用字节流好点（1byte = 8位）；字符流： 它处理的单元为2个字节的Unicode字符，分别操作字符、字符数组或字符串，字符流是由Java虚拟机将字节转化为2个字节的Unicode字符为单位的字符而成的，如果是关系到中文（文本）的，用字符流好点（1Unicode = 2字节 = 16位）；所有文件的储存是都是字节（byte）的储存，在磁盘上保留的并不是文件的字符而是先把字符编码成字节，再储存这些字节到磁盘。在读取文件（特别是文本文件）时，也是一个字节一个字节地读取以形成字节序列。字节流可用于任何类型的对象，包括二进制对象，而字符流只能处理字符或者字符串；字节流提供了处理任何类型的IO操作的功能，但它不能直接处理Unicode字符，而字符流就可以。字节流是最基本的，所有的InputStrem和OutputStream的子类都是,主要用在处理二进制数据，它是按字节来处理的 但实际中很多的数据是文本，又提出了字符流的概念，它是按虚拟机的encode来处理，也就是要进行字符集的转化 这两个之间通过 InputStreamReader,OutputStreamWriter来关联，实际上是通过byte[]和String来关联。在实际开发中出现的汉字问题实际上都是在字符流和字节流之间转化不统一而造成的。最简单的区分字节流和字符流：万物皆文件，那就将文件在记事本里面打开，如果打开后能看的懂的就是字符流，如果看不懂那就是字节流。问题:word.doc 数据字节流还是字符流？答：.doc数据字节流。Excel 数据字节流还是字符流？答：要根据保存的格式进行判断，如果是保存为.csv那么他就是字符流，如果是其他的则数据字节流。字符和字节操作如何简单记住字节流和字符流字符流是以Reader，Writer结尾的字节流是以InputStream或OutputStream结尾的2、字节流的使用字节流主要是操作byte类型数据，以byte数组为准，主要操作类就是OutputStream、InputStream。字节输出流：OutputStream，OutputStream是整个IO包中字节输出流的最大父类Closeable表示可以关闭的操作2.1普通字节流的操作import java.io.FileInputStream; import java.io.FileNotFoundException; import java.io.FileOutputStream; import java.io.IOException; /** * 复制D:\io_test下的加菲猫文件到D:\io_test这个目录下 */ public class IoDemo4 { public static void main(String[] args) throws IOException { //输入文件目录 String srcFilePath = "D:\\io_test\\加菲猫.jpg"; //输出文件目录（拷贝的路径） String destFilePath = "D:\\io_test\\加菲猫2.jpg"; FileInputStream fileInputStream = null; FileOutputStream fileOutputStream = null; try { //输入流,传入路径 fileInputStream = new FileInputStream(srcFilePath); //输出流 fileOutputStream = new FileOutputStream(destFilePath); byte[] bytes = new byte[1024]; try { int count = 0; //如果不等于-1说明还没有读取完成，要继续读取 while((count = fileInputStream.read(bytes)) != -1){ //还有内容要读取,然后从0开始读取，上面count有读了多少个，那么就写多少个 fileOutputStream.write(bytes,0,count); } } catch (IOException e) { e.printStackTrace(); } } catch (FileNotFoundException e) { e.printStackTrace(); }finally { //用完之后一定要关闭流,但是得判断不是空的时候才需要去关闭流 if(fileInputStream != null){ fileInputStream.close(); } if(fileOutputStream != null){ fileOutputStream.close(); } } } }注意一定要关闭流：因为每天计算机打开的文件的数量是有限的，以liunx为例，最大可以打开66533个文件，但是无论执行结果怎么样他都要关闭流，那么就需要将他放在finally里面，所有上面的声明也得放在外面如果传入的地址和目标地址都是错的，也就是找不到，如果传入的地址错了，那么还没有初始化就已经被调用关闭流了，所以要进行判断2.2带有缓冲区的字节流的操作什么是缓存区？定义：缓存区相当于缓存，它是存在内存中的写操作：没有使用缓存区：CPU读取每个字节之后直接操作磁盘（性能比较底）进行写完，写操作的瓶颈就会出现，因为每个字节都会操作一次磁盘使用缓冲区：那么每次会将字符放入缓存区（内存），等缓冲区满了之后，才一次性写入磁盘因为内存的操作速度远远大于磁盘，因此带缓冲区的输入流和输出流实现的效率就非常高（比如扔垃圾，一次性扔完和一次次扔肯定消耗的时间是有很大差距的）import java.io.BufferedInputStream; import java.io.BufferedOutputStream; import java.io.FileInputStream; import java.io.FileOutputStream; /** * 带有缓冲区的字节流操作:图片复制 * */ public class IoDemo5 { public static void main(String[] args) { //输入文件目录 String srcFilePath = "D:\\io_test\\加菲猫.jpg"; //输出文件目录（拷贝的路径） String destFilePath = "D:\\io_test\\加菲猫3.jpg"; try { //因为带有缓存区的是基于原始的类进行操作的 BufferedInputStream bufferedInputStream = new BufferedInputStream(new FileInputStream(srcFilePath)); BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(new FileOutputStream(destFilePath)); { byte[] bytes = new byte[1024]; int count = 0; while ((count = bufferedInputStream.read(bytes)) != -1){ bufferedOutputStream.write(bytes,0,count); } } }catch (Exception e){ e.printStackTrace(); } } }3、字符流的使用3.1普通的字符流操作字符输出流：Writer字符输入流：Reader3.1.1内容的写入import java.io.FileWriter; import java.io.IOException; /** * 内容的写入：D:\io_test\1下创建一个text.txt文件，并写入内容 */ public class IoDemo7 { public static void main(String[] args) throws IOException { //定义文件地址 String filePath = "D:\\io_test\\1\\text.txt"; //写入内容 String content = "欢迎来到Java~"; //因为是写入文件，所以要使用FileWriter方法 FileWriter fileWriter = null; try { //因为这里的参数append会被默认成false，所以如果是多行写入的话需要手动设置成true fileWriter = new FileWriter(filePath,false); fileWriter.write(content); } catch (IOException e) { e.printStackTrace(); }finally { //关闭流 fileWriter.close(); } } }注意：如果不关闭流，虽然也能创建文件成功，不报错，但是可能存在内容无法写入的情况3.1.2将文件下的内容读取到控制台import java.io.File; import java.io.FileNotFoundException; import java.util.Scanner; /** * 读取文件内容 */ public class IoDemo8 { public static void main(String[] args) throws FileNotFoundException { String filePath = "D:\\io_test\\1\\tt.txt"; Scanner scanner = new Scanner(new File(filePath)); //因为读取的内容可能不止一条，所以加上while循环 while (scanner.hasNext()){ System.out.println(scanner.nextLine()); } } }3.1.3将一个路径下的内容复制到另一个路径下import java.io.FileReader; import java.io.FileWriter; import java.io.IOException; /** *将一个文件的内容写入到另一个文件下 * 如：D:\io_test\1\text.txt -> text2.txt */ public class IoDemo9 { public static void main(String[] args) throws IOException { //读取文件的地址 String srcFilePath = "D:\\io_test\\1\\text.txt"; //目标文件的地址 String destFilePath = "D:\\io_test\\1\\text2.txt"; FileReader fileReader = null; FileWriter fileWriter = null; //先读后写 try{ fileReader = new FileReader(srcFilePath); fileWriter = new FileWriter(destFilePath); { //读操作 char[] chars = new char[1024]; while (true){ int count = fileReader.read(chars); //判断是否等于-1，如果等于-1，说明已经读取完了，跳出循环 if(count == -1){ break; } //写操作 fileWriter.write(chars,0,count); } } }catch (Exception e){ e.printStackTrace(); }finally { fileReader.close(); fileWriter.close(); } } }3.2带有缓冲区的字符流操作内容的复制import java.io.*; /** * 带有缓冲区的字符流操作 * 如：内容的复制 */ public class IoDemo10 { public static void main(String[] args) throws IOException { String srcFilePath = "D:\\io_test\\1\\tt.txt"; String destFilePath = "D:\\io_test\\1\\tt2.txt"; BufferedReader bufferedReader = null; BufferedWriter bufferedWriter = null; //先读取，后写入 try { bufferedReader = new BufferedReader(new FileReader(srcFilePath)); bufferedWriter = new BufferedWriter(new FileWriter(destFilePath)); { char[] chars = new char[1024]; while (true){ int count = bufferedReader.read(chars); if(count == -1){ break; } bufferedWriter.write(chars,0,count); } } }catch (Exception e){ e.printStackTrace(); }finally { bufferedReader.close(); bufferedWriter.close(); } } }4、总结字节流操作的基本单元是字节；字符流操作的基本单元为Unicode码元。字节流在操作的时候本身不会用到缓冲区的，是与文件本身直接操作的；而字符流在操作的时候使用到缓冲区的。所有文件的存储都是字节(byte)的存储，在磁盘上保留的是字节。在使用字节流操作中，即使没有关闭资源（close方法），也能输出；而字符流不使用close方法的话，不会输出任何内容
1、消息应答的方法：Channel.basicAck(用于肯定确认)，RabbitMQ已知道该消息并且成功处理，可以将其丢弃Channel.basicNack(用于否定确认)Channel.basicReject(用于否定确认)，与Channel.basicNack相比少了一个参数，不处理该消息了，直接拒绝，可以将其丢弃了。2、Multiple的解释true表示批量应答channel上未应答的消息，比如channel上有传送tag的消息5,6,7,8,，当前tag是8，那么此时5-8的这些还未应答的消息就会被确认收到消息应答false同上面相比只会应答tag=8的消息，5,6,7这三个消息依然不会被确认收到消息应答手动应答的好处是可以批量应答并且减少网络拥堵 。3、消息手动应答代码//getEnvelope()属性，getDeliveryTag()标记 /* * 参数1：消息的标记 tag * 参数2：是否批量应答，false：不批量应答；true：批量 * */ channel.basicAck(message.getEnvelope().getDeliveryTag(), false); // 采用手动应答 boolean autoAck = false; channel.basicConsume(QUEUE_NAME,autoAck,deliverCallback,(consumerTag) -> { System.out.println(consumerTag+"消费者取消消费接口回调逻辑"); });4、持久化// 声明队列 // 持久化 需要让Queue持久化 boolean durable = true; channel.queueDeclare(TASK_QUEUE_NAME,durable,false,false,null);注意：如果之前声明的队列不是持久化的，需要把原先队列先删除，或者重新创建一个持久化的队列，不然就会出现错误。将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉 RabbitMQ 将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强，但是对于我们的简单任务队列而言，这已经绰绰有余了。5、不公平分发//不公平分发channel //basicQos(0) --> 轮询分发（公平分发），默认值为0 //channel.basicQos(1) --> 不公平分发，能者多劳 //channel.basicQos(>2) --> 预取值 int prefetchCount = 1; channel.basicQos(prefetchCount); //采用手动应答 boolean autoAck = false; channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);6、发布确认发布确认的策略:开启发布确认的方法：发布确认默认是没有开启的，如果要开启，需要调用方法 confirmSelect，每当你要想使用发布确认，都需要在 channel 上调用该方法。//开启发布确认 channel.confirmSelect();6.1单个确认发布这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布，waitForConfirmsOrDie(long) 这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。这种确认方式有一个最大的缺点就是：发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。/** * 单个发送 */ public static void publishMessageIndividually() throws Exception { Channel channel = RabbitMqUtils.getChannel(); //队列声明 String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName, true, false, false, null); //开启发布确认 channel.confirmSelect(); long begin = System.currentTimeMillis(); for (int i = 0; i < MESSAGE_COUNT; i++) { String message = i + ""; channel.basicPublish("", queueName, null, message.getBytes()); //服务端返回 false 或超时时间内未返回，生产者可以消息重发 //单个消息马上进行发布确认 boolean flag = channel.waitForConfirms(); if (flag) { System.out.println("消息发送成功"); } } long end = System.currentTimeMillis(); System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息,耗时" + (end - begin) + "ms"); }6.2批量确认发布上面那种方式非常慢，与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量。缺点就是：当发生故障导致发布出现问题时，不知道是哪个消息出问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布。/** * 批量确认发送 */ public static void publishMessageBatch() throws Exception { Channel channel = RabbitMqUtils.getChannel(); //队列声明 String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName, true, false, false, null); //开启发布确认 channel.confirmSelect(); //批量确认消息大小 int batchSize = 100; //未确认消息个数 int outstandingMessageCount = 0; long begin = System.currentTimeMillis(); for (int i = 0; i < MESSAGE_COUNT; i++) { String message = i + ""; channel.basicPublish("", queueName, null, message.getBytes()); outstandingMessageCount++; if (outstandingMessageCount == batchSize) { channel.waitForConfirms(); outstandingMessageCount = 0; } } //为了确保还有剩余没有确认消息 再次确认 if (outstandingMessageCount > 0) { channel.waitForConfirms(); } long end = System.currentTimeMillis(); System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息,耗时" + (end - begin) + "ms"); }6.3异步确认发布异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说， 它是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功。/** * 异步确认发布 * */ public static void publishMessageByAsync() throws Exception { Channel channel = RabbitMqUtils.getChannel(); String queueName = UUID.randomUUID().toString(); // 生成一个队列 channel.queueDeclare(queueName, false, false, false, null); // 开启发布确认 channel.confirmSelect(); long begin = System.currentTimeMillis(); // 消息确认成功回调 /** * 参数1。消息的标记 * 参数2.是否为批量确认 */ ConfirmCallback ackCallback = (deliveryTag, multiple) -> { System.out.println(String.format("确认的消息：" + deliveryTag)); }; // 消息确认失败回调 /** * 参数1。消息的标记 * 参数2.是否为批量确认 */ ConfirmCallback nackCallback = (deliveryTag, multiple) -> { System.out.println(String.format("未确认的消息：" + deliveryTag)); }; // 添加异步监听器 监听消息是否成功 /** * 参数1。监听哪些消息成功了 * 参数2.监听哪些消息失败了 */ channel.addConfirmListener(ackCallback, nackCallback); for (int i = 0; i < MESSAGE_COUNT; i++) { String message = i + ""; channel.basicPublish("", queueName, null, message.getBytes()); } long end = System.currentTimeMillis(); System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) + "ms"); }处理异步未确认消息：最好的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列， 比如说用 ConcurrentLinkedQueue 这个队列在 confirm callbacks 与发布线程之间进行消息的传递。/** * 异步确认 * 消息确认失败时的处理 * @throws Exception */ public static void publishMessageByAsyncNackCallback() throws Exception { Channel channel = RabbitMqUtils.getChannel(); String queueName = UUID.randomUUID().toString(); // 生成一个队列 channel.queueDeclare(queueName, false, false, false, null); // 开启发布确认 channel.confirmSelect(); /** * 线程安全有序的一个哈希表，适用于高并发的情况下 高并发下的hashMap * 1.轻松的将序号与消息进行关联 * 2.可以轻松的批量的删除条目，只需要给到序号 * 3.支持高并发（多线程） */ ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>(); // 消息确认成功回调 /** * 参数1。消息的标记 * 参数2.是否为批量确认 */ ConfirmCallback ackCallback = (deliveryTag, multiple) -> { //判断是否进行批量操作 if (multiple) { //2.删除已经确认的消息 剩下的就是确认失败的消息 //headMap标记在头部 ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag); confirmed.clear(); } else { outstandingConfirms.remove(deliveryTag); } System.out.println("确认的消息：" + deliveryTag); }; // 消息确认失败回调 /** * 参数1。消息的标记 * 参数2.是否为批量确认 */ ConfirmCallback nackCallback = (deliveryTag, multiple) -> { //3.打印一下未确认的消息 String message = outstandingConfirms.get(deliveryTag); System.out.println("未确认的消息是：" + message + "，未确认消息的tag是：" + deliveryTag); }; // 添加异步监听器 监听消息是否成功 /** * 参数1。监听哪些消息成功了 * 参数2.监听哪些消息失败了 */ channel.addConfirmListener(ackCallback, nackCallback); //开始时间 long begin = System.currentTimeMillis(); for (int i = 0; i < MESSAGE_COUNT; i++) { String message = i + ""; channel.basicPublish("", queueName, null, message.getBytes()); //1.记录所有要发送的消息 //getNextPublishSeqNo下一次发布的序号 outstandingConfirms.put(channel.getNextPublishSeqNo(), message); } long end = System.currentTimeMillis(); System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) + "ms"); }完整代码：package com.youzi.rabbitmq.four; import com.rabbitmq.client.Channel; import com.rabbitmq.client.ConfirmCallback; import com.youzi.rabbitmq.utils.RabbitMqUtils; import java.util.UUID; import java.util.concurrent.ConcurrentNavigableMap; import java.util.concurrent.ConcurrentSkipListMap; /** * @Author youzi * @ClassName ConfirmMessage * @Date 2022/7/4 10:34 * @Description //TODO */ public class ConfirmMessage { public static final int MESSAGE_COUNT = 1000; public static void main(String[] args) throws Exception { //单个发送 22240ms ConfirmMessage.publishMessageIndividually(); //批量确认 311ms ConfirmMessage.publishMessageBatch(); //异步确认 51ms ConfirmMessage.publishMessageByAsync(); //异步确认加失败处理 49ms ConfirmMessage.publishMessageByAsyncNackCallback(); } /** * 单个发送 * * @throws Exception */ public static void publishMessageIndividually() throws Exception { Channel channel = RabbitMqUtils.getChannel(); String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName, true, false, false, null); //开启发布确认 channel.confirmSelect(); long begin = System.currentTimeMillis(); for (int i = 0; i < MESSAGE_COUNT; i++) { String message = i + ""; channel.basicPublish("", queueName, null, message.getBytes()); //服务端返回 false 或超时时间内未返回，生产者可以消息重发 //单个消息马上进行发布确认 boolean flag = channel.waitForConfirms(); if (flag) { System.out.println("消息发送成功"); } } long end = System.currentTimeMillis(); System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息,耗时" + (end - begin) + "ms"); } /** * 批量确认 * * @throws Exception */ public static void publishMessageBatch() throws Exception { Channel channel = RabbitMqUtils.getChannel(); String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName, true, false, false, null); //开启发布确认 channel.confirmSelect(); long begin = System.currentTimeMillis(); //批量确认消息大小 int bachSize = 100; //批量发布消息 批量确认 for (int i = 0; i < MESSAGE_COUNT; i++) { String message = i + ""; channel.basicPublish("", queueName, null, message.getBytes()); //判断达到100条消息的时候，批量确认一次 if (i % bachSize == 0) { //发布确认 channel.waitForConfirms(); } } long end = System.currentTimeMillis(); System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息,耗时" + (end - begin) + "ms"); } /** * 异步确认 * * @throws Exception */ public static void publishMessageByAsync() throws Exception { Channel channel = RabbitMqUtils.getChannel(); String queueName = UUID.randomUUID().toString(); // 生成一个队列 channel.queueDeclare(queueName, false, false, false, null); // 开启发布确认 channel.confirmSelect(); long begin = System.currentTimeMillis(); // 消息确认成功回调 /** * 参数1。消息的标记 * 参数2.是否为批量确认 */ ConfirmCallback ackCallback = (deliveryTag, multiple) -> { System.out.println(String.format("确认的消息：" + deliveryTag)); }; // 消息确认失败回调 /** * 参数1。消息的标记 * 参数2.是否为批量确认 */ ConfirmCallback nackCallback = (deliveryTag, multiple) -> { System.out.println(String.format("未确认的消息：" + deliveryTag)); }; // 添加异步监听器 监听消息是否成功 /** * 参数1。监听哪些消息成功了 * 参数2.监听哪些消息失败了 */ channel.addConfirmListener(ackCallback, nackCallback); for (int i = 0; i < MESSAGE_COUNT; i++) { String message = i + ""; channel.basicPublish("", queueName, null, message.getBytes()); } long end = System.currentTimeMillis(); System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) + "ms"); } /** * 异步确认 * 消息确认失败时的处理 * @throws Exception */ public static void publishMessageByAsyncNackCallback() throws Exception { Channel channel = RabbitMqUtils.getChannel(); String queueName = UUID.randomUUID().toString(); // 生成一个队列 channel.queueDeclare(queueName, false, false, false, null); // 开启发布确认 channel.confirmSelect(); /** * 线程安全有序的一个哈希表，适用于高并发的情况下 高并发下的hashMap * 1.轻松的将序号与消息进行关联 * 2.可以轻松的批量的删除条目，只需要给到序号 * 3.支持高并发（多线程） */ ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>(); // 消息确认成功回调 /** * 参数1。消息的标记 * 参数2.是否为批量确认 */ ConfirmCallback ackCallback = (deliveryTag, multiple) -> { //判断是否进行批量操作 if (multiple) { //2.删除已经确认的消息 剩下的就是确认失败的消息 //headMap标记在头部 ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag); confirmed.clear(); } else { outstandingConfirms.remove(deliveryTag); } System.out.println("确认的消息：" + deliveryTag); }; // 消息确认失败回调 /** * 参数1。消息的标记 * 参数2.是否为批量确认 */ ConfirmCallback nackCallback = (deliveryTag, multiple) -> { //3.打印一下未确认的消息 String message = outstandingConfirms.get(deliveryTag); System.out.println("未确认的消息是：" + message + "，未确认消息的tag是：" + deliveryTag); }; // 添加异步监听器 监听消息是否成功 /** * 参数1。监听哪些消息成功了 * 参数2.监听哪些消息失败了 */ channel.addConfirmListener(ackCallback, nackCallback); //开始时间 long begin = System.currentTimeMillis(); for (int i = 0; i < MESSAGE_COUNT; i++) { String message = i + ""; channel.basicPublish("", queueName, null, message.getBytes()); //1.记录所有要发送的消息 //getNextPublishSeqNo下一次发布的序号 outstandingConfirms.put(channel.getNextPublishSeqNo(), message); } long end = System.currentTimeMillis(); System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) + "ms"); } }以上三种发布确认对比发布确认方案对比单独发布确认消息同步等待确认，简单，但吞吐量非常有限批量发布确认消息批量同步等待，简单，合理的吞吐量，出现问题无法确认是哪条消息异常异步发布确认消息最佳性能和资源使用，可以异步处理成功和失败的消息7、交换机7.1、Fanout模式（发布/订阅模式）概念：它是将接收到的所有消息广播到它知道的所有队列中。生产者代码：package com.youzi.rabbitmq.five; import com.rabbitmq.client.Channel; import com.youzi.rabbitmq.utils.RabbitMqUtils; import java.util.Scanner; /** * @Author youzi * @ClassName Emitlog * @Date 2022/7/4 16:15 * @Description //TODO */ public class Emitlog { //交换机名称 public static final String EXCHANGE_NAME = "logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //声明一个交换机 channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); Scanner scanner = new Scanner(System.in); while (scanner.hasNext()) { String message = scanner.next(); channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8")); System.out.println("生产者发出消息：" + message); } } }消费者1代码：package com.youzi.rabbitmq.five; import com.rabbitmq.client.Channel; import com.rabbitmq.client.ConfirmCallback; import com.rabbitmq.client.DeliverCallback; import com.youzi.rabbitmq.utils.RabbitMqUtils; import com.youzi.rabbitmq.utils.SleepUtils; /** * @Author youzi * @ClassName ReceiveLogs01 * @Date 2022/7/4 16:15 * @Description //TODO 消费者1 */ public class ReceiveLogs01 { //交换机名称 public static final String EXCHANGE_NAME = "logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //声明一个交换机 channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); //声明一个临时队列 /** * 生成一个临时队列,队列的名称是随机的 * 当消费者断开与队列的连接时，队列就自动删除了 */ String queueName = channel.queueDeclare().getQueue(); //绑定交换机与队列 /** * 参数1.队列名称 * 参数2.交换机名称 * 参数3.routingKey值 */ channel.queueBind(queueName, EXCHANGE_NAME, ""); System.out.println("等待接收消息，把接收到的消息放在屏幕上。。。。。。"); DeliverCallback deliverCallback = (consumerTag, message) -> { System.out.println("ReceiveLogs01接收到的消息：" + new String(message.getBody())); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); } }消费者2代码：package com.youzi.rabbitmq.five; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import com.youzi.rabbitmq.utils.RabbitMqUtils; /** * @Author youzi * @ClassName ReceiveLogs02 * @Date 2022/7/4 16:15 * @Description //TODO */ public class ReceiveLogs02 { //交换机名称 public static final String EXCHANGE_NAME = "logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //声明一个交换机 channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); //声明一个临时队列 /** * 生成一个临时队列,队列的名称是随机的 * 当消费者断开与队列的连接时，队列就自动删除了 */ String queueName = channel.queueDeclare().getQueue(); //绑定交换机与队列 /** * 参数1.队列名称 * 参数2.交换机名称 * 参数3.routingKey值 */ channel.queueBind(queueName, EXCHANGE_NAME, ""); System.out.println("等待接收消息，把接收到的消息放在屏幕上。。。。。。"); DeliverCallback deliverCallback = (consumerTag, message) -> { System.out.println("ReceiveLogs02接收到的消息：" + new String(message.getBody())); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); } } 7.2、Direct 模式（路由模式）概念：消息只去到它绑定的 routingKey 队列中去。生产者代码：package com.youzi.rabbitmq.six; import com.rabbitmq.client.Channel; import com.youzi.rabbitmq.utils.RabbitMqUtils; import java.util.Scanner; /** * @Author youzi * @ClassName DirectLogs * @Date 2022/7/4 17:23 * @Description //TODO */ public class DirectLogs { private static final String EXCHANGE_NAME = "direct_logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); Scanner scanner = new Scanner(System.in); while (scanner.hasNext()) { String message = scanner.next(); //第二个参数可以填入info，warning，error，填入哪个就是哪个消费者接收 channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8")); System.out.println("生产者发出消息：" + message); } } }消费者1代码：package com.youzi.rabbitmq.six; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import com.youzi.rabbitmq.utils.RabbitMqUtils; /** * @Author youzi * @ClassName ReceiveLogsDirect01 * @Date 2022/7/4 17:22 * @Description //TODO */ public class ReceiveLogsDirect01 { private static final String EXCHANGE_NAME = "direct_logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); //队列声明 channel.queueDeclare("console", false, false, false, null); //队列绑定 channel.queueBind("console", EXCHANGE_NAME, "info"); channel.queueBind("console", EXCHANGE_NAME, "warning"); //发送回调 DeliverCallback deliverCallback = (consumerTag, message) -> { System.out.println("ReceiveLogsDirect01接收到的消息：" + new String(message.getBody())); }; channel.basicConsume("console", true, deliverCallback, consumerTag -> { }); } }消费者2代码：package com.youzi.rabbitmq.six; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import com.youzi.rabbitmq.utils.RabbitMqUtils; /** * @Author youzi * @ClassName ReceiveLogsDirect02 * @Date 2022/7/4 17:22 * @Description //TODO */ public class ReceiveLogsDirect02 { private static final String EXCHANGE_NAME = "direct_logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); //队列声明 channel.queueDeclare("disk", false, false, false, null); //队列绑定 channel.queueBind("disk", EXCHANGE_NAME, "error"); //发送回调 DeliverCallback deliverCallback = (consumerTag, message) -> { System.out.println("ReceiveLogsDirect02接收到的消息：" + new String(message.getBody())); }; channel.basicConsume("disk", true, deliverCallback, consumerTag -> { }); } }7.3、Topic 模式（主题模式）主题交换机是通过匹配路由键进行消息的转发，对于key匹配上的，转发相应队列，全都不匹配的直接丢弃；规则：所有的单词均是由 . 分隔；可以匹配任意一个单词；可以匹配0个或多个单词；注意：当一个队列的routeKey是#，那么这个队列将接收所有消息，类似于FANOUT;当一个队列的routeKey没有包含#和*,那么这个队列类似于DIRECT;routeKey匹配*.hello.*匹配三个单词，且中间的必须是hellowelcome.*匹配两个单词，第一个必须是welcomehello.world匹配两个单词，第一个是hello，第二个是world#.happy匹配至少一个单词，且最后一个单词必须是happyhappy.*.#匹配至少两个单词，第一个必须是happy生产者代码：package com.youzi.rabbitmq.seven; import com.rabbitmq.client.Channel; import com.youzi.rabbitmq.utils.RabbitMqUtils; import java.util.HashMap; import java.util.Map; public class EmitLogTopic { //交换机的名称 public static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); /** * Q1-->绑定的是 * 中间带 orange 带 3 个单词的字符串(*.orange.*) * Q2-->绑定的是 * 最后一个单词是 rabbit 的 3 个单词(*.*.rabbit) * 第一个单词是 lazy 的多个单词(lazy.#) */ HashMap<String, String> bindingKeyMap = new HashMap<>(); bindingKeyMap.put("quick.orange.rabbit", "被队列 Q1Q2 接收到"); bindingKeyMap.put("lazy.orange.elephant", "被队列 Q1Q2 接收到"); bindingKeyMap.put("quick.orange.fox", "被队列 Q1 接收到"); bindingKeyMap.put("lazy.brown.fox", "被队列 Q2 接收到"); bindingKeyMap.put("lazy.pink.rabbit", "虽然满足两个绑定但只被队列 Q2 接收一次"); bindingKeyMap.put("quick.brown.fox", "不匹配任何绑定不会被任何队列接收到会被丢弃"); bindingKeyMap.put("quick.orange.male.rabbit", "是四个单词不匹配任何绑定会被丢弃"); bindingKeyMap.put("lazy.orange.male.rabbit", "是四个单词但匹配 Q2"); for (Map.Entry<String, String> bindingKeyEntry : bindingKeyMap.entrySet()) { String bindingKey = bindingKeyEntry.getKey(); String message = bindingKeyEntry.getValue(); channel.basicPublish(EXCHANGE_NAME, bindingKey, null, message.getBytes("UTF-8")); System.out.println("生产者发出消息：" + message); } } }消费者代码1：package com.youzi.rabbitmq.seven; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import com.youzi.rabbitmq.utils.RabbitMqUtils; /** * @Author youzi * @ClassName ReceiveLogsTopic01 * @Date 2022/7/5 14:11 * @Description //TODO */ public class ReceiveLogsTopic01 { //交换机名称 private static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //声明交换机 channel.exchangeDeclare(EXCHANGE_NAME, "topic"); //声明队列 String queueName = "Q1"; channel.queueDeclare(queueName, false, false, false, null); channel.queueBind(queueName, EXCHANGE_NAME, "*.orange.*"); System.out.println("等待接收消息...."); //发送回调 DeliverCallback deliverCallback = (consumerTag, message) -> { System.out.println("ReceiveLogsTopic01接收到的消息：" + new String(message.getBody())); System.out.println("接收队列：" + queueName + "绑定的键：" + message.getEnvelope().getRoutingKey()); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); } }消费者代码2：package com.youzi.rabbitmq.seven; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import com.youzi.rabbitmq.utils.RabbitMqUtils; /** * @Author youzi * @ClassName ReceiveLogsTopic01 * @Date 2022/7/5 14:11 * @Description //TODO */ public class ReceiveLogsTopic02 { //交换机名称 private static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //声明交换机 channel.exchangeDeclare(EXCHANGE_NAME, "topic"); //声明队列 String queueName = "Q2"; channel.queueDeclare(queueName, false, false, false, null); channel.queueBind(queueName, EXCHANGE_NAME, "*.*.rabbit"); channel.queueBind(queueName, EXCHANGE_NAME, "lazy.#"); System.out.println("等待接收消息...."); //发送回调 DeliverCallback deliverCallback = (consumerTag, message) -> { System.out.println("ReceiveLogsTopic02接收到的消息：" + new String(message.getBody())); System.out.println("接收队列：" + queueName + "绑定的键：" + message.getEnvelope().getRoutingKey()); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); } }8、死信队列死信就是无法被消费的消息；当生产者把消息投递到broker或者直接到队列中，消费者从队列中取出消息进行消费，但由于特殊情况，队列中的个别消息无法被消费，这样的消息如果没有后续的处理，就会变成死信，存放在死信队列中；通常进入死信队列的消息要么是异常消息，要么是超时消息；死信队列的来源：消息TTL时间过期；队列达到最大长度，无法添加新的消息到mq中；消息被拒绝（basic.reject或者nack）并且requeue=false;8.1、消息TTL时间过期场景模拟：声明两个直接交换机和队列，逻辑交换机绑定逻辑队列，死信交换机绑定死信队列，然后生产者发送消息到逻辑交换机，由于逻辑交换机接收超时，导致消息进入死信队列，被消费者2处理；声明TTL时间过期在Consumer中可以queueDeclare队列声明处增加参数map，使用keyx-message-ttl设置；（不推荐）在生产者发送消息的时候，通过BasicProperties设置发送消息的TTL，推荐使用，不同消息可以设置不同的消息TTL；生产者代码：package com.youzi.rabbitmq.eight; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.youzi.rabbitmq.utils.RabbitMqUtils; /** * @Author youzi * @ClassName Producer * @Date 2022/7/11 14:27 * @Description //TODO */ public class Producer { //正常的交换机 public static final String NORMAL_EXCHANGE = "normal_exchange"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //死信消息 设置TTL时间（过期时间） 单位是ms AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build(); //发送消息 for (int i = 1; i < 11; i++) { String message = "info" + i; channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes()); } } }消费者1代码：package com.youzi.rabbitmq.eight; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import com.youzi.rabbitmq.utils.RabbitMqUtils; import java.util.HashMap; import java.util.Map; /** * @Author youzi * @ClassName Consumer01 * @Date 2022/7/11 10:53 * @Description //TODO 死信队列 */ public class Consumer01 { //正常的交换机 public static final String NORMAL_EXCHANGE = "normal_exchange"; //死信的交换机 public static final String DEAD_EXCHANGE = "dead_exchange"; //正常的队列 public static final String NORMAL_QUEUE = "normal_queue"; //死信的交换机 public static final String DEAD_QUEUE = "dead_queue"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //声明正常和死信交换机 channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT); //声明正常队列 Map<String, Object> arg = new HashMap<>(); //过期时间 单位ms 10s = 10000 //(两个地方可以设置，建议在生产者设置，因为消费者这边设置的是固定值，不可更改，消费者可以设置每条消息的过期时间) arg.put("x-message-ttl", 10000); //正常交换机设置死信交换机 arg.put("x-dead-letter-exchange", DEAD_EXCHANGE); //设置死信RoutingKey arg.put("x-dead-letter-routing-key", "lisi"); channel.queueDeclare(NORMAL_QUEUE, false, false, false, arg); //声明死信队列 channel.queueDeclare(DEAD_QUEUE, false, false, false, null); //普通队列绑定普通交换机 channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan"); //死信队列绑定死信交换机 channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi"); System.out.println("等待接收消息....."); //发送回调 DeliverCallback deliverCallback = (consumerTag, message) -> { System.out.println("Consumer01接受的消息是：" + new String(message.getBody())); }; channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, consumerTag -> { }); } }消费者2代码：package com.youzi.rabbitmq.eight; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import com.youzi.rabbitmq.utils.RabbitMqUtils; import java.util.HashMap; import java.util.Map; /** * @Author youzi * @ClassName Consumer01 * @Date 2022/7/11 10:53 * @Description //TODO 死信队列 */ public class Consumer02 { //死信的交换机 public static final String DEAD_QUEUE = "dead_queue"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); System.out.println("等待接收消息....."); //发送回调 DeliverCallback deliverCallback = (consumerTag, message) -> { System.out.println("Consumer02接受的消息是：" + new String(message.getBody())); }; channel.basicConsume(DEAD_QUEUE, true, deliverCallback, consumerTag -> { }); } }8.2、队列达到最大长度，无法添加新的消息到MQ操作：在声明logic逻辑队列的参数中，增加key值为x-max-length；标识队列最大消息长度；先要把原来的normal_queue队列删除，否则启动会报错，代码与上面的差不多，只需要把生产者的超时时间删除// 在消费者1的代码中，参数map增加设置超时时间的key即可 // 声明队列最大长度6条消息 arg.put("x-max-length", 6);8.3、消息被拒绝消费者1需要手动确认消息，把确认消息设置成reject；则该消息进入死信队列；消费者1代码：package com.youzi.rabbitmq.eight; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import com.youzi.rabbitmq.utils.RabbitMqUtils; import java.util.HashMap; import java.util.Map; /** * @Author youzi * @ClassName Consumer01 * @Date 2022/7/11 10:53 * @Description //TODO 死信队列 */ public class Consumer01 { //正常的交换机 public static final String NORMAL_EXCHANGE = "normal_exchange"; //死信的交换机 public static final String DEAD_EXCHANGE = "dead_exchange"; //正常的队列 public static final String NORMAL_QUEUE = "normal_queue"; //死信的交换机 public static final String DEAD_QUEUE = "dead_queue"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //声明正常和死信交换机 channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT); //声明正常队列 Map<String, Object> arg = new HashMap<>(); //过期时间 单位ms 10s = 10000 //(两个地方可以设置，建议在生产者设置，因为消费者这边设置的是固定值，不可更改，消费者可以设置每条消息的过期时间) //arg.put("x-message-ttl", 10000); //正常交换机设置死信交换机 arg.put("x-dead-letter-exchange", DEAD_EXCHANGE); //设置死信RoutingKey arg.put("x-dead-letter-routing-key", "lisi"); //声明队列最大长度6条消息 //arg.put("x-max-length", 6); channel.queueDeclare(NORMAL_QUEUE, false, false, false, arg); //声明死信队列 channel.queueDeclare(DEAD_QUEUE, false, false, false, null); //普通队列绑定普通交换机 channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan"); //死信队列绑定死信交换机 channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi"); System.out.println("等待接收消息....."); //发送回调 DeliverCallback deliverCallback = (consumerTag, message) -> { String msg = new String(message.getBody()); if (msg.equals("info5")) { System.out.println("Consumer01接受的消息是：" + msg + ":此消息是被C1拒绝的"); //false表示不塞回队列，即进入死信队列 channel.basicReject(message.getEnvelope().getDeliveryTag(), false); } else { System.out.println("Consumer01接受的消息是：" + msg); //false指的是不批量应答 channel.basicAck(message.getEnvelope().getDeliveryTag(), false); } }; //开启手动应答 channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, consumerTag -> { }); } }消费者2和生产者代码不变
1.Docker安装RabbitMQ1.1下载RabbitMQ镜像docker pull rabbitmq1.2创建并启动RabbitMQ容器docker run -id --hostname myrabbit --name rabbitmq -p 15672:15672 -p 5672:5672 rabbitmq1.3进入容器交互页面docker exec -it rabbitmq /bin/bash1.4下载插件rabbitmq-plugins enable rabbitmq_management1.5登录ip+端口号访问账号密码均为：guest1.6拓展# 创建账号和密码 rabbitmqctl add_user 用户名 密码 # 设置用户角色 rabbitmqctl set_user_tags admin administrator # 为用户添加资源权限 # set_permissions [-p <vhostpath>] <user> <conf> <write> <read> # 添加配置、写、读权限 rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"用户级别：administrator：可以登录控制台、查看所有信息、可以对 rabbitmq 进行管理。monitoring：监控者 登录控制台，查看所有信息。policymaker：策略制定者 登录控制台，指定策略。managment：普通管理员 登录控制台。注：有可能会出现下面这个问题访问Docker的rabbitMq容器的Web-UI界面无法查看消息的状态及出现Stats in management UI are disabled on this node解决办法：进入RabbitMQ容器docker exec -it rabbitmq /bin/bash进入容器后，cd到路径/etc/rabbitmq/conf.d/cd /etc/rabbitmq/conf.d/再修改management_agent.disable_metrics_collectorecho management_agent.disable_metrics_collector = false > management_agent.disable_metrics_collector.conf退出容器exit输入指令docker exec -it rabbitmq rabbitmq-plugins enable rabbitmq_management重启RabbitMQdocker restart rabbitmq2.简单使用2.1pom.xml文件添加依赖(本环境是jdk1.8的)<dependencies> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.8.0</version> </dependency> <dependency> <groupId>commons-io</groupId> <artifactId>commons-io</artifactId> <version>2.6</version> </dependency> </dependencies>2.2新建Producer类为生产者package com.youzi.rabbitmq.one; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * @Author youzi * @ClassName Producer * @Date 2022/6/30 15:41 * @Description //TODO 生产者 */ public class Producer { //设置队列名称 public static final String QUEUE_NAME = "hello"; public static void main(String[] args) throws IOException, TimeoutException { //1.创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); //2.设置IP factory.setHost("ip地址"); //3.设置用户名密码 factory.setUsername("用户名"); factory.setPassword("密码"); //4.创建连接 Connection connection = factory.newConnection(); //5.获取信道 Channel channel = connection.createChannel(); /** * 参数1.队列名称 * 参数2.队列里面的消息是否持久化，默认是不持久化 * 参数3.该队列是否只供一个消费之进行消费，是否进行消息共享，true表示可以多个消费者消费 * 参数4.是否自动删除，最后一个消费者断开连接以后，该队列是否自动删除，true表示自动删除 * 参数5.其他参数 */ //生成一个队列 channel.queueDeclare(QUEUE_NAME,false,false,false,null); //消息 String message = "hello youzi"; /** * 参数1.交换机名称 * 参数2.路由的key值，队列名称 * 参数3.其他参数 * 参数4.消息体 */ channel.basicPublish("",QUEUE_NAME,null,message.getBytes()); System.out.println("消息发送成功"); channel.close(); connection.close(); } }2.3新建Consumer类为消费者package com.youzi.rabbitmq.one; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * @Author youzi * @ClassName Consumer * @Date 2022/6/30 16:44 * @Description //TODO 消费者 */ public class Consumer { //设置队列名称 一定要和生产者的一样才能找到对应的关系 public static final String QUEUE_NAME = "hello"; public static void main(String[] args) throws IOException, TimeoutException { //1.创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); //2.设置IP factory.setHost("IP地址"); //3.设置用户名密码 factory.setUsername("用户名"); factory.setPassword("密码"); //4.创建连接 Connection connection = factory.newConnection(); //5.获取信道 Channel channel = connection.createChannel(); //申明回调方法 未成功消费 DeliverCallback deliverCallback = (consumerTag, message) -> { System.out.println(new String(message.getBody())); }; //申明回调方法 取消消费 CancelCallback cancelCallback = consumerTag -> { System.out.println("消费中断"); }; /** * 消费者接收消息 * 参数1.队列名称 * 参数2.消费成功之后是否要自动应答，true表示自动应答 * 参数3.消费者未成功消费的回调方法 * 参数4.消费者取消消费时的回调方法 */ channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback); } }2.4访问RabbitMQ的后台管理页面
首先<dependencies> <dependency> <groupId>joda-time</groupId> <artifactId>joda-time</artifactId> <version>2.10.1</version> </dependency> </dependencies>其次import org.joda.time.Interval; import org.joda.time.Period; import java.text.DateFormat; import java.text.ParseException; import java.text.SimpleDateFormat; import java.util.Calendar; import java.util.Date; import java.util.HashMap; import java.util.Map; /** * @Author youzi * @ClassName TestDateUtil * startDate 开始时间 * endDate 结束时间 * dateFormat 时间格式（形如 yyyy-MM-dd） * @return 相差的月份数和天数 months 、 days * @throws ParseException * @Date 2022/6/22 14:02 * @Description 计算两个时间差 */ public class TestDateUtil { public static void main(String[] args) throws ParseException { DateFormat df = new SimpleDateFormat("yyyy-MM-dd"); Date dateStart = df.parse("2022-03-04"); Date dateEnd = df.parse("2022-06-03"); if (dateStart.getTime() > dateEnd.getTime()) { Date temp = dateStart; dateStart = dateEnd; dateEnd = temp; } Calendar c = Calendar.getInstance(); c.setTime(dateEnd); c.add(Calendar.DAY_OF_MONTH, 1); //这是明天 dateEnd = c.getTime(); Interval interval = new Interval(dateStart.getTime(), dateEnd.getTime()); Period p = interval.toPeriod(); Integer years = p.getYears(); Integer months = p.getMonths(); Integer weeks = p.getWeeks(); Integer days = p.getDays(); Integer resultMonths = years * 12 + months; Integer resultDays = weeks * 7 + days; Map<String, Integer> result = new HashMap<>(); result.put("months", resultMonths); result.put("days", resultDays); System.out.println(result); } }