首页 文章 精选 留言 我的

精选列表

搜索[源码学习]，共10000篇文章
优秀的个人博客,低调大师

Spring Cloud限流详解（内含源码）

原文：http://www.itmuch.com/spring-cloud-sum/spring-cloud-ratelimit/转载声明：本博客由周立创作，采用CC BY 3.0 CN许可协议。可自由转载、引用，但需署名作者且注明文章出处。如转载至微信公众号，请在文末添加作者公众号二维码。 在高并发的应用中，限流往往是一个绕不开的话题。本文详细探讨在Spring Cloud中如何实现限流。 在Zuul 上实现限流是个不错的选择，只需要编写一个过滤器就可以了，关键在于如何实现限流的算法。常见的限流算法有漏桶算法以及令牌桶算法。这个可参考 https://www.cnblogs.com/LBSer/p/4083131.html ，写得通俗易懂，你值得拥有，我就不拽文了。 Google Guava 为我们提供了限流工具类RateLimiter ，于是乎，我们可以撸代码了。 代码示例 @Component public class RateLimitZuulFilter extends ZuulFilter { private final RateLimiter rateLimiter = RateLimiter.create(1000.0); @Override public String filterType() { return FilterConstants.PRE_TYPE; } @Override public int filterOrder() { return Ordered.HIGHEST_PRECEDENCE; } @Override public boolean shouldFilter() { // 这里可以考虑弄个限流开启的开关，开启限流返回true，关闭限流返回false，你懂的。 return true; } @Override public Object run() { try { RequestContext currentContext = RequestContext.getCurrentContext(); HttpServletResponse response = currentContext.getResponse(); if (!rateLimiter.tryAcquire()) { HttpStatus httpStatus = HttpStatus.TOO_MANY_REQUESTS; response.setContentType(MediaType.TEXT_PLAIN_VALUE); response.setStatus(httpStatus.value()); response.getWriter().append(httpStatus.getReasonPhrase()); currentContext.setSendZuulResponse(false); throw new ZuulException( httpStatus.getReasonPhrase(), httpStatus.value(), httpStatus.getReasonPhrase() ); } } catch (Exception e) { ReflectionUtils.rethrowRuntimeException(e); } return null; } } 如上，我们编写了一个pre 类型的过滤器。对Zuul过滤器有疑问的可参考我的博客： Spring Cloud内置的Zuul过滤器详解：http://www.itmuch.com/spring-cloud/zuul/zuul-filter-in-spring-cloud Spring Cloud Zuul过滤器详解：http://www.itmuch.com/spring-cloud/zuul/spring-cloud-zuul-filter 在过滤器中，我们使用Guava RateLimiter 实现限流，如果已经达到最大流量，就抛异常。 分布式场景下的限流 以上单节点Zuul下的限流，但在生产中，我们往往会有多个Zuul实例。对于这种场景如何限流呢？我们可以借助Redis实现限流。 使用redis实现，存储两个key，一个用于计时，一个用于计数。请求每调用一次，计数器增加1，若在计时器时间内计数器未超过阈值，则可以处理任务 if(!cacheDao.hasKey(TIME_KEY)) { cacheDao.putToValue(TIME_KEY, 0, 1, TimeUnit.SECONDS); } if(cacheDao.hasKey(TIME_KEY) && cacheDao.incrBy(COUNTER_KEY, 1) > 400) { // 抛个异常什么的 } 实现微服务级别的限流 一些场景下，我们可能还需要实现微服务粒度的限流。此时可以有两种方案： 方式一：在微服务本身实现限流。 和在Zuul上实现限流类似，只需编写一个过滤器或者拦截器即可，比较简单，不作赘述。个人不太喜欢这种方式，因为每个微服务都得编码，感觉成本很高啊。 加班那么多，作为程序猿的我们，应该学会偷懒，这样才可能有时间孝顺父母、抱老婆、逗儿子、遛狗养鸟、聊天打屁、追求人生信仰。好了不扯淡了，看方法二吧。 方法二：在Zuul上实现微服务粒度的限流。 在讲解之前，我们不妨模拟两个路由规则，两种路由规则分别代表Zuul的两种路由方式。 zuul: routes: microservice-provider-user: /user/** user2: url: http://localhost:8000/ path: /user2/** 如配置所示，在这里，我们定义了两个路由规则，microservice-provider-user 以及user2 ，其中microservice-provider-user 这个路由规则使用到Ribbon + Hystrix，走的是RibbonRoutingFilter ；而user2 这个路由用不上Ribbon也用不上Hystrix，走的是SipleRoutingFilter 。如果你搞不清楚这点，请参阅我的博客： Spring Cloud内置的Zuul过滤器详解：http://www.itmuch.com/spring-cloud/zuul/zuul-filter-in-spring-cloud Spring Cloud Zuul过滤器详解：http://www.itmuch.com/spring-cloud/zuul/spring-cloud-zuul-filter 搞清楚这点之后，我们就可以撸代码了： @Component public class RateLimitZuulFilter extends ZuulFilter { private Map<String, RateLimiter> map = Maps.newConcurrentMap(); @Override public String filterType() { return FilterConstants.PRE_TYPE; } @Override public int filterOrder() { // 这边的order一定要大于org.springframework.cloud.netflix.zuul.filters.pre.PreDecorationFilter的order // 也就是要大于5 // 否则，RequestContext.getCurrentContext()里拿不到serviceId等数据。 return Ordered.LOWEST_PRECEDENCE; } @Override public boolean shouldFilter() { // 这里可以考虑弄个限流开启的开关，开启限流返回true，关闭限流返回false，你懂的。 return true; } @Override public Object run() { try { RequestContext context = RequestContext.getCurrentContext(); HttpServletResponse response = context.getResponse(); String key = null; // 对于service格式的路由，走RibbonRoutingFilter String serviceId = (String) context.get(SERVICE_ID_KEY); if (serviceId != null) { key = serviceId; map.putIfAbsent(serviceId, RateLimiter.create(1000.0)); } // 如果压根不走RibbonRoutingFilter，则认为是URL格式的路由 else { // 对于URL格式的路由，走SimpleHostRoutingFilter URL routeHost = context.getRouteHost(); if (routeHost != null) { String url = routeHost.toString(); key = url; map.putIfAbsent(url, RateLimiter.create(2000.0)); } } RateLimiter rateLimiter = map.get(key); if (!rateLimiter.tryAcquire()) { HttpStatus httpStatus = HttpStatus.TOO_MANY_REQUESTS; response.setContentType(MediaType.TEXT_PLAIN_VALUE); response.setStatus(httpStatus.value()); response.getWriter().append(httpStatus.getReasonPhrase()); context.setSendZuulResponse(false); throw new ZuulException( httpStatus.getReasonPhrase(), httpStatus.value(), httpStatus.getReasonPhrase() ); } } catch (Exception e) { ReflectionUtils.rethrowRuntimeException(e); } return null; } } 简单讲解一下这段代码： 对于microservice-provider-user 这个路由，我们可以用context.get(SERVICE_ID_KEY); 获取到serviceId，获取出来就是microservice-provider-user ； 而对于user2 这个路由，我们使用context.get(SERVICE_ID_KEY); 获得是null，但是呢，可以用context.getRouteHost() 获得路由到的地址，获取出来就是http://localhost:8000/ 。接下来的事情，你们懂的。 改进与提升 实际项目中，除以上实现的限流方式，还可能会： 一、在上文的基础上，增加配置项，控制每个路由的限流指标，并实现动态刷新，从而实现更加灵活的管理 二、基于CPU、内存、数据库等压力限流（感谢平安常浩智）提出。。 下面，笔者借助Spring Boot Actuator提供的Metrics 能力进行实现基于内存压力的限流——当可用内存低于某个阈值就开启限流，否则不开启限流。 @Component public class RateLimitZuulFilter extends ZuulFilter { @Autowired private SystemPublicMetrics systemPublicMetrics; @Override public boolean shouldFilter() { // 这里可以考虑弄个限流开启的开关，开启限流返回true，关闭限流返回false，你懂的。 Collection<Metric<?>> metrics = systemPublicMetrics.metrics(); Optional<Metric<?>> freeMemoryMetric = metrics.stream() .filter(t -> "mem.free".equals(t.getName())) .findFirst(); // 如果不存在这个指标，稳妥起见，返回true，开启限流 if (!freeMemoryMetric.isPresent()) { return true; } long freeMemory = freeMemoryMetric.get() .getValue() .longValue(); // 如果可用内存小于1000000KB，开启流控 return freeMemory < 1000000L; } // 省略其他方法 } 三、实现不同维度的限流，例如： 对请求的目标URL进行限流（例如：某个URL每分钟只允许调用多少次） 对客户端的访问IP进行限流（例如：某个IP每分钟只允许请求多少次） 对某些特定用户或者用户组进行限流（例如：非VIP用户限制每分钟只允许调用100次某个API等） 多维度混合的限流。此时，就需要实现一些限流规则的编排机制。与、或、非等关系。 参考文档 分布式环境下限流方案的实现：http://blog.csdn.net/Justnow_/article/details/53055299
2017-11-30
优秀的个人博客,低调大师

关于hive RegexSerDe的源码分析

最近有个业务建表使用了 RegexSerDe，之前虽然也它来解析nginx日志，但是没有做深入的了解。这次看了下其实现方式。 建表语句： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 CREATEexternalTABLEods_cart_log ( time_localSTRING, request_jsonSTRING, trace_id_numSTRING ) PARTITIONEDBY ( dtstring, hourstring ) ROWFORMATSERDE 'org.apache.hadoop.hive.contrib.serde2.RegexSerDe' WITHSERDEPROPERTIES ( "input.regex" = "\\\[(.*?)\\\].*\\\|(.*?)(.*?)\\\[(.*?)\\\]" , "output.format.string" = "%1$s%2$s%4$s" ) STOREDASTEXTFILE; 测试数据： 1 2 3 4 5 [2014-07-2415:54:54][6]OperationData.php: :89|{ "action" : "add" , "redis_key_hash" :9, "time" : "1406188494.73745500" , "source" : "web" , "mars_cid" : "" , "session_id" : "" , "info" :{ "cart_id" :26885, "user_id" :4, "size_id" : "2784145" , "num" : "1" , "warehouse" : "VIP_NH" , "brand_id" : "7379" , "cart_record_id" :26885, "channel" : "te" }} trace_id[40618849399972881308] 这里trace_id_num按照猜想应该是第4个字段（即40618849399972881308），但是实际输出了第3个字段（trace_id） 查看其代码实现： RegexSerDe主要由下面三个参数： 1）input.regex 正则 2）output.format.string 输出格式 3）input.regex.case.insensitive 大小写是否敏感 其中input.regex用在反序列化方法中，即数据的读取（hive读取hdfs文件），相对的output.format.string 用在序列化的方法中，即数据的写入（hive写入hdfs文件）。 在反序列化的方法deserialize中有如下代码，用于返回代表匹配字段的数据： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 for ( int c= 0 ;c<numColumns;c++){ //numColumns是按表中column的数量算的（ 比如这个例子columnNames是[time_local,request_json,trace_id_num]|numColumns=columnNames.size(); try { row.set(c,m.group(c+ 1 )); //可以看到字段的匹配从0开始，中间不会有跳跃， 所以这里selecttrace_id_num字段是正则里面的第 3 个组，而和output.format.string没有关系 } catch (RuntimeExceptione){ partialMatchedRows++; if (partialMatchedRows>=nextPartialMatchedRows){ nextPartialMatchedRows=getNextNumberToDisplay(nextPartialMatchedRows); //Reporttherow LOG.warn( "" +partialMatchedRows + "partiallyunmatchedrowsarefound," + "cannotfindgroup" +c+ ":" +rowText); } row.set(c, null ); } } work around的方法有两个，1个是把所有正则匹配的字段列出，另一个就是更改正则的分组，只拿自己care的分组，比如上面可以改为 1 \\\[(.*?)\\\].*\\\|(.*?).*?\\\[(.*?)\\\] 这里output.format.string的设置仔细想想貌似没什么用，首先RegexSerDe的方式只在textfile下生效，即可以用load向hive的表中导入数据，但是load是一个hdfs层面的文件操作，不涉及到序列化，如果想使用序列化，需要使用insert into select的方式插入数据，但是这种方式插入的数据又和select的数据有关系，和output.format.string没什么关系了。。 其实regexserde类有两个 分别位于 1 ./serde/src/java/org/apache/hadoop/hive/serde2/RegexSerDe.java和 1 ./contrib/src/java/org/apache/hadoop/hive/contrib/serde2/RegexSerDe.java 都是扩展了AbstractSerDe这个抽象类。通过代码可以看到contrib下的这个类是实现了serialize和deserialize方法，而上面这个只实现了deserialize方法，由此看来RegexSerDe中的serialize方法可能是没什么用的。。 另外需要注意几点： 1.如果一行匹配不上，整个行的字段输出都是null 1 2 3 4 5 6 7 8 9 if (!m.matches()){ unmatchedRows++; if (unmatchedRows>=nextUnmatchedRows){ nextUnmatchedRows=getNextNumberToDisplay(nextUnmatchedRows); //Reporttherow LOG.warn( "" +unmatchedRows+ "unmatchedrowsarefound:" +rowText); } return null ; } 2.表的字段类型必须都是string，否则会报错,如果需要别的字段，可以在select中使用cast做转换 1 2 3 4 5 6 7 for ( int c= 0 ;c<numColumns;c++){ if (!columnTypes.get(c).equals(TypeInfoFactory.stringTypeInfo)){ throw new SerDeException(getClass().getName() + "onlyacceptsstringcolumns,butcolumn[" +c+ "]named" +columnNames.get(c)+ "hastype" +columnTypes.get(c)); } } 本文转自菜菜光 51CTO博客，原文链接：http://blog.51cto.com/caiguangguang/1532987，如需转载请自行联系原作者
2017-11-29
优秀的个人博客,低调大师

hiverc文件的加载源码实现

使用过hive的都知道，可以通过指定 -i参数或者配置.hiverc来设置hive启动时初始执行的一些命令，比如可以把udf的定义写到.hiverc文件中。加载.hiverc的过程是在CliDriver类中定义的。 具体的方法调用顺序： 1 main--->run--->executeDriver----->processInitFiles---->processFile---->processReader---->processLine--->processCmd，processCmd里面的具体实现就比较多了。 1）其中executeDriver中，处理 初始化文件的调用部分： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 //Execute-iinitfiles(alwaysinsilentmode) cli.processInitFiles(ss); //加载初始文件 if (ss.execString!= null ){ int cmdProcessStatus=cli.processLine(ss.execString); //-e的情况ss.execString=commandLine.getOptionValue('e'); return cmdProcessStatus; } try { if (ss.fileName!= null ){ return cli.processFile(ss.fileName); //-f的情况ss.fileName=commandLine.getOptionValue('f'); } } catch (FileNotFoundExceptione){ System.err.println( "Couldnotopeninputfileforreading.(" +e.getMessage()+ ")" ); return 3 ; } 2）其中在processReader定义了.hiverc文件中注释的写法（以--开头） 1 2 3 4 5 6 7 8 9 10 11 public int processReader(BufferedReaderr) throws IOException{ Stringline; StringBuilderqsb= new StringBuilder(); while ((line=r.readLine())!= null ){ //Skippingthroughcomments if (!line.startsWith( "--" )){ qsb.append(line+ "\n" ); } } return (processLine(qsb.toString())); } 3）processInitFiles定义了加载的顺序： -i指定或者 1 $HIVE_HOME/bin/.hiverc--->$HIVE_CONF_DIR/.hiverc--->${user.home}/.hiverc 代码实现： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 public void processInitFiles(CliSessionStatess) throws IOException{ boolean saveSilent=ss.getIsSilent(); ss.setIsSilent( true ); for (StringinitFile:ss.initFiles){ //ss.initFiles指由hive-i参数传入的文件路径 int rc=processFile(initFile); if (rc!= 0 ){ System.exit(rc); } } if (ss.initFiles.size()== 0 ){ //如果没有指定-i参数，则按如下路径加载 if (System.getenv( "HIVE_HOME" )!= null ){ //$HIVE_HOME/bin/.hiverc StringhivercDefault=System.getenv( "HIVE_HOME" )+File.separator+ "bin" +File.separator+HIVERCFILE; if ( new File(hivercDefault).exists()){ int rc=processFile(hivercDefault); if (rc!= 0 ){ System.exit(rc); } console.printError( "Puttingtheglobalhivercin" + "$HIVE_HOME/bin/.hivercisdeprecated.Please" + "use$HIVE_CONF_DIR/.hivercinstead." ); } } if (System.getenv( "HIVE_CONF_DIR" )!= null ){ //$HIVE_CONF_DIR/.hiverc StringhivercDefault=System.getenv( "HIVE_CONF_DIR" )+File.separator +HIVERCFILE; if ( new File(hivercDefault).exists()){ int rc=processFile(hivercDefault); if (rc!= 0 ){ System.exit(rc); } } } if (System.getProperty( "user.home" )!= null ){ //${user.home}/.hiverc StringhivercUser=System.getProperty( "user.home" )+File.separator+ HIVERCFILE; if ( new File(hivercUser).exists()){ int rc=processFile(hivercUser); if (rc!= 0 ){ System.exit(rc); } } } } ss.setIsSilent(saveSilent); } 本文转自菜菜光 51CTO博客，原文链接：http://blog.51cto.com/caiguangguang/1542269，如需转载请自行联系原作者
2017-11-21

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

时间: 2026-04-11 大小: 211.28KB 下载: 56次
Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

时间: 2026-09-24 大小: 189MB 下载: 9次
Spring

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

时间: 2026-08-21 大小: 5MB 下载: 2次
Rocky Linux

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

时间: 2026-08-21 大小: 1.42GB 下载: 6次
热门标签
DeepSeek Jdk25 OpenAi HarmonyOS6 Nacos3 SpringBoot4 SpringCloud Docker Kubernetes Service Mesh Redis Gemini3 Rocky

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册