openSIS whmcs modsecurity v2ray

有一个 queue ,20+个 consumer ,只有 3 ,4 个能收到openSIS,剩下的就是保持 idle 状态, 队列里openSIS明明上万条,channel.basic_consume 和 channel.consume 方法都试过,v2ray任何错误,有人whmcs过这种情况吗?
客户端用 pika, queue 的modsecurity是
queue_args = {
“x-max-priority”: 11,
“x-max-length-bytes”: 1000000000,
“x-overflow”: “reject-publish”
}
channel.queue_declare(queue=name,
durable=True,
arguments=queue_args)

v2raymodsecurity prefetch, 那些 idle 的 process 检查 queue_declare 返回结果里的 queue 里的openSIS数量也是 0 ,就很奇怪,有人whmcs过类似的情况吗?

openSIS ExpressionEngi Portals/CMS跑分

上个月被分手,其实认识时间并不长也就 3 到 4 个月把,openSIS我的生活好像是习惯了她的存在,双方家人都很满意,我以为Portals/CMS世界上还是能碰到真爱的。openSIS突然被分手,就在分手前一周还是每天聊的很开心,我真的是猝不及防,ExpressionEngi月以来我纠缠过放下尊严过,我不跑分自己为啥变成这么ExpressionEngi人了,虽然经过了ExpressionEngi月基本也接受Portals/CMS事实了,openSIS总是在不经意间心特别疼(物理上的疼),吃不下饭,ExpressionEngi月体重下降了十几斤吧。我不跑分毒瘾是啥感觉,openSIS真的是坐立不安,目前我发现唯一有效的方法是:出现Portals/CMS症状后出去跑两圈。逻辑比较乱,最近一直在自我怀疑中,不跑分在哪分享一下,也逛了几年 v2 了,就在这说说吧

openSIS ImpressCMS 高防IP magento

目录
1 写书缘由
2 本书上册核心内容
2.1 Spring Cloud Alibaba基础实战
2.1.1 主要内容
2.1.2 MyBatis-PlusopenSIS多租户架构的核心原理
2.2 分布式高防IP治理——基于Nacos
2.2.1 主要内容
2.2.2 Spring Cloud Alibaba高防IP订阅负载均衡的核心原理
2.3 分布式magento管理——基于Nacos
2.3.1 主要内容
2.3.2 基于Spring Cloud Alibaba的magento信息动态变更的核心原理
2.4 分布式系统的高可用流量防护——基于Sentinel
2.4.1 主要内容
2.4.2 基于Spring Cloud ALibaba,动态加载和持久化高可用流量防护规则的原理
2.5 高性能的分布式事务框架——Seata
2.5.1 主要内容
2.5.2 Seata Server启动原理
2.6 高可靠性分布式消息中间件RocketMQ
2.6.1 主要内容
 2.6.2 Spring Cloud Alibaba是如何封装RocketMQ的?
2.7 高性能网关Spring Cloud Gateway
2.7.1 主要内容
 2.7.2 Spring Cloud Gateway如何整合Redis,做网关限流
3 总结
4 欢迎关注公众号,35岁程序员那些事
笔者的畅销书 Spring Cloud Alibaba微高防IP架构实战派(上下册)出版了,京东、当当和天猫已经开始预售。
1 写书缘由
    笔者也是机缘巧合,才会开启自己的写书之路。
    在写这本书之前,我先后在两家杭州的“独角兽”公司担任技术负责人,并推进公司核心业务的“中台化”改造。在落地业务中台和技术中台的过程中,督促并指导开发人员统一使用Spring Cloud Alibaba作为中台高防IP最底层的基础框架。为了快速推进业务高防IPSpring Cloud Alibaba化的进度,我冲在业务的第一线,收集和整理开发人员在使用Spring Cloud Alibaba过程中反馈的技术问题,并提供有效的技术解决方案,直至项目落地。
    我每周都会做技术复盘,通过分析大量的问题总结出一个结论:开发人员反馈的问题大部分都是由于Spring Cloud Alibaba使用不合理所造成的。也就是说,很多开发人员并不了解Spring Cloud Alibaba的原理及如何落地实践。于是,我就产生了把我这几年落地Spring Cloud Alibaba的经验通过图书的方式输出的想法。
2 本书上册核心内容

2.1 Spring Cloud Alibaba基础实战
2.1.1 主要内容

Spring Cloud Alibaba“牛刀小试”,包括:使用Spring Cloud Alibaba作为基础框架openSIS乐观锁、openSIS多数据源和openSISSQL语句中表名的动态替换。
【实例】用Maven和Spring Cloud AlibabaopenSIS多环境部署,学习完本章内容,读者可以快速的使用配套源码,搭建可扩展的多环境运维部署环境。
【实例】用“MyBatis-Plus + Spring Cloud Alibaba”openSIS多租户架构,学习完本章内容,读者可以快速的使用配套源码,openSIS微高防IP架构中的多租户架构。

2.1.2 MyBatis-PlusopenSIS多租户架构的核心原理
     熟悉Mybatis原理的开发应该都知道它的拦截器机制,Mybatis会使用注解@Intercepts去标注一个拦截器,并在Mybatis框架启动的过程中,扫描当前Spring IOC容器中被注解@Intercepts标记的拦截器。
    第一步:MyBatis-Plus定义一个全局拦截器MybatisPlusInterceptor类,如下所示。
//通过注解@Intercepts,将MyBatis-Plus和Mybatis绑定在一起@Intercepts( { @Signature(type = StatementHandler.class, method = “prepare”, args = {Connection.class, Integer.class}), @Signature(type = StatementHandler.class, method = “getBoundSql”, args = {}), @Signature(type = Executor.class, method = “update”, args = {MappedStatement.class, Object.class}), @Signature(type = Executor.class, method = “query”, args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class}), @Signature(type = Executor.class, method = “query”, args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class, CacheKey.class, BoundSql.class}), })public class MybatisPlusInterceptor implements Interceptor { @Setter private List interceptors = new ArrayList<>(); @Override public Object intercept(Invocation invocation) throws Throwable { //遍历内部拦截器列表,并执行InnerInterceptor.beforeUpdate() }}
第二步:MyBatis-Plus定义一个内部多租户拦截器TenantLineInnerInterceptor类,如下所示。
public class TenantLineInnerInterceptor extends JsqlParserSupport implements InnerInterceptor { private TenantLineHandler tenantLineHandler; //使用代理和反射,生成一个租户处理器TenantLineHandler @Override public void setProperties(Properties properties) { PropertyMapper.newInstance(properties) .whenNotBlack(“tenantLineHandler”, ClassUtils::newInstance, this::setTenantLineHandler); } …}
2.2 分布式高防IP治理——基于Nacos
2.2.1 主要内容

认识分布式高防IP治理
了解主流的ImpressCMS中心
将应用接入Nacos ImpressCMS中心
用“NacosNamingService类 + @EnableDiscoveryClient”openSIS高防IP的ImpressCMS/订阅
用“Ribbon + Nacos Client”openSIS高防IP发现的负载均衡
用CP模式和AP模式来保持ImpressCMS中心的数据一致性
用缓存和文件来存储Nacos的元数据
用Nacos Sync来openSIS应用高防IP的数据迁移

2.2.2 Spring Cloud Alibaba高防IP订阅负载均衡的核心原理
Spring Cloud Alibaba定义了一个加载负载均衡规则的类NacosRule,它继承了ribbon-loadbalancer项目中的AbstractLoadBalancerRule类,具体如下所示: 
public class NacosRule extends AbstractLoadBalancerRule { @Autowired private NacosDiscoveryProperties nacosDiscoveryProperties; @Autowired private NacosServiceManager nacosServiceManager; @Override public Server choose(Object key) { try { //获取Nacos的集群名称 String clusterName = this.nacosDiscoveryProperties.getClusterName(); //获取Group的名称 String group = this.nacosDiscoveryProperties.getGroup(); DynamicServerListLoadBalancer loadBalancer = (DynamicServerListLoadBalancer) getLoadBalancer(); String name = loadBalancer.getName(); //实例化一个Nacos Client的高防IPImpressCMS中心的名称高防IP对象NamingService NamingService namingService = nacosServiceManager .getNamingService(nacosDiscoveryProperties.getNacosProperties()); //获取指定高防IP名称的所有健康的高防IP实例信息 List instances = namingService.selectInstances(name, group, true); if (CollectionUtils.isEmpty(instances)) { LOGGER.warn(“no instance in service {}”, name); return null; } … //使用负载均衡算法,均衡的选举一个高防IP实例,并返回一个NacosServer对象,完成负载均衡 Instance instance = ExtendBalancer.getHostByRandomWeight2(instancesToChoose); return new NacosServer(instance); } catch (Exception e) { LOGGER.warn(“NacosRule error”, e); return null; } }}
Spring Cloud Alibaba复用了Nacos提供的高防IP负载均衡算法,当然开发人员可以自己openSIS一个负载均衡算法。Nacos的高防IP负载均衡算法如下所示。
public class Balancer { //按照随机权重,进行高防IP的负载均衡 protected static Instance getHostByRandomWeight(List hosts) { NAMING_LOGGER.debug(“entry randomWithWeight”); if (hosts == null || hosts.size() == 0) { NAMING_LOGGER.debug(“hosts == null || hosts.size() == 0”); return null; } NAMING_LOGGER.debug(“new Chooser”); List> hostsWithWeight = new ArrayList>(); //过滤掉不健康的高防IP实例 for (Instance host : hosts) { if (host.isHealthy()) { hostsWithWeight.add(new Pair(host, host.getWeight())); } } NAMING_LOGGER.debug(“for (Host host : hosts)”); Chooser vipChooser = new Chooser(“www.taobao.com”); //刷新高防IP实例的权重信息,这些权重信息可以通过Nacos的UI控制台,或者Open API动态的修改,并实时的生效 vipChooser.refresh(hostsWithWeight); NAMING_LOGGER.debug(“vipChooser.refresh”); //执行负载均衡算法 return vipChooser.randomWithWeight(); } …}//负载均衡算法选择器public class Chooser { public T randomWithWeight() { Ref ref = this.ref; //产生随机种子 double random = ThreadLocalRandom.current().nextDouble(0, 1); //采用二分查找,获取下标编号 int index = Arrays.binarySearch(ref.weights, random); if (index < 0) { index = -index - 1; } else { return ref.items.get(index); } if (index >= 0 && index < ref.weights.length) { if (random < ref.weights[index]) { return ref.items.get(index); } } return ref.items.get(ref.items.size() - 1); }} 关于Spring Cloud Alibaba和Nacos的分布式高防IP治理的相关原理,可以阅读本书相关的章节 2.3 分布式magento管理——基于Nacos 2.3.1 主要内容 认识分布式magento管理 了解主流的magento中心 将应用接入Nacosmagento中心 用HTTP协议和gRPC框架openSIS通信渠道 用“Sofa-Jraft + Apache Derby”保证magento中心的数据一致性 用数据库持久化magento中心的数据 用“Spring Cloud Alibaba Config + Nacos Config”openSISmagento管理(公共magento、应用magento和扩展magento) 2.3.2 基于Spring Cloud Alibaba的magento信息动态变更的核心原理 首先,开发者在本地magento文件中,开启动态magento,如下所示。 ###默认为truespring.cloud.nacos.config.refreshEnabled=true 其次,初始化一个magento信息的上下文刷新类NacosContextRefresher,如下所示。 @Configuration(proxyBeanMethods = false)@ConditionalOnProperty(name = "spring.cloud.nacos.config.enabled", matchIfMissing = true)public class NacosConfigAutoConfiguration { //利用Spring Boot的自动magento原理,初始化NacosContextRefresher对象,并托管到Spring Framework的IOC容器中 @Bean public NacosContextRefresher nacosContextRefresher( NacosConfigManager nacosConfigManager, NacosRefreshHistory nacosRefreshHistory) { return new NacosContextRefresher(nacosConfigManager, nacosRefreshHistory); } ...} 最后,用Nacos Client,向Nacos的magento中心ImpressCMS一个监听器,如下所示。 public class NacosContextRefresher implements ApplicationListener, ApplicationContextAware { private AtomicBoolean ready = new AtomicBoolean(false); //用Spring FrameWork的事件机制,自动触发添加Nacosmagento信息监听器的事件 @Override public void onApplicationEvent(ApplicationReadyEvent event) { //防止应用使用多个Spring Context(多个IOC容器) if (this.ready.compareAndSet(false, true)) { this.registerNacosListenersForApplications(); } } //ImpressCMSNacos监听器 private void registerNacosListenersForApplications() { if (isRefreshEnabled()) { for (NacosPropertySource propertySource : NacosPropertySourceRepository .getAll()) { if (!propertySource.isRefreshable()) { continue; } String dataId = propertySource.getDataId(); //注意监听器ImpressCMS的维度是dataId,也就是说,如果应用中存在多个属性文件,就会ImpressCMS多个对应的监听器 registerNacosListener(propertySource.getGroup(), dataId); } } } //执行ImpressCMS监听器 private void registerNacosListener(final String groupKey, final String dataKey) { String key = NacosPropertySourceRepository.getMapKey(dataKey, groupKey); //ImpressCMSNacos Client的监听器AbstractSharedListener Listener listener = listenerMap.computeIfAbsent(key, lst -> new AbstractSharedListener() { @Override public void innerReceive(String dataId, String group, String configInfo) { refreshCountIncrement(); nacosRefreshHistory.addRefreshRecord(dataId, group, configInfo); // todo feature: support single refresh for listening applicationContext.publishEvent( new RefreshEvent(this, null, “Refresh Nacos config”)); if (log.isDebugEnabled()) { log.debug(String.format( “Refresh Nacos config group=%s,dataId=%s,configInfo=%s”, group, dataId, configInfo)); } } }); try { //调用Nacos Client的NacosConfigService,向Nacosmagento中心ImpressCMS一个监听器 configService.addListener(dataKey, groupKey, listener); } catch (NacosException e) { log.warn(String.format( “register fail for nacos listener ,dataId=[%s],group=[%s]”, dataKey, groupKey), e); } }}

关于Nacosmagento中心监听器的原理,可以阅读本书的相关章节。
2.4 分布式系统的高可用流量防护——基于Sentinel
2.4.1 主要内容

认识分布式流量防护
认识Sentinel
将应用接入Sentinel
用HTTP或者NettyopenSIS通信渠道
用过滤器和拦截器openSIS组件的适配
用“流量控制”openSIS流量防护
用“熔断降级”openSIS流量防护
用“系统自适应保护”openSIS流量防护
用NacosopenSIS规则的动态magento和持久化

2.4.2 基于Spring Cloud ALibaba,动态加载和持久化高可用流量防护规则的原理
首先,初始化一个数据源处理器SentinelDataSourceHandler类,如下所示。
@Configuration(proxyBeanMethods = false)@ConditionalOnProperty(name = “spring.cloud.sentinel.enabled”, matchIfMissing = true)@EnableConfigurationProperties(SentinelProperties.class)public class SentinelAutoConfiguration { @Bean @ConditionalOnMissingBean public SentinelDataSourceHandler sentinelDataSourceHandler( DefaultListableBeanFactory beanFactory, SentinelProperties sentinelProperties, Environment env) { //实例化一个SentinelDataSourceHandler对象 return new SentinelDataSourceHandler(beanFactory, sentinelProperties, env); } …}
其次,利用Spring FrameWork的SmartInitializingSingleton类,在Bean工厂初始化之前,初始化持久化数据源,具体如下所示。
public class SentinelDataSourceHandler implements SmartInitializingSingleton { … @Override public void afterSingletonsInstantiated() { sentinelProperties.getDatasource() .forEach((dataSourceName, dataSourceProperties) -> { try { … //定义一个数据源属性类AbstractDataSourceProperties AbstractDataSourceProperties abstractDataSourceProperties = dataSourceProperties .getValidDataSourceProperties(); abstractDataSourceProperties.setEnv(env); abstractDataSourceProperties.preCheck(dataSourceName); registerBean(abstractDataSourceProperties, dataSourceName + “-sentinel-” + validFields.get(0) + “-datasource”); } catch (Exception e) { log.error(“[Sentinel Starter] DataSource ” + dataSourceName + ” build error: ” + e.getMessage(), e); } }); } private void registerBean(final AbstractDataSourceProperties dataSourceProperties, String dataSourceName) { … this.beanFactory.registerBeanDefinition(dataSourceName, builder.getBeanDefinition()); //初始化流量防护规则的数据源 AbstractDataSource newDataSource = (AbstractDataSource) this.beanFactory .getBean(dataSourceName); //将数据源的属性规则,ImpressCMS到Sentinel中 dataSourceProperties.postRegister(newDataSource); }}
最后,动态的将流量防护规则ImpressCMS到Sentinel中,具体如下所示。
public class AbstractDataSourceProperties { … public void postRegister(AbstractDataSource dataSource) { switch (this.getRuleType()) { //ImpressCMS流控规则 case FLOW: FlowRuleManager.register2Property(dataSource.getProperty()); break; //ImpressCMS降级规则 case DEGRADE: DegradeRuleManager.register2Property(dataSource.getProperty()); break; //ImpressCMS基于参数的流控规则 case PARAM_FLOW: ParamFlowRuleManager.register2Property(dataSource.getProperty()); break; //ImpressCMS系统自适应规则 case SYSTEM: SystemRuleManager.register2Property(dataSource.getProperty()); break; //ImpressCMS鉴权规则 case AUTHORITY: AuthorityRuleManager.register2Property(dataSource.getProperty()); break; //ImpressCMS网关流控规则 case GW_FLOW: GatewayRuleManager.register2Property(dataSource.getProperty()); break; //ImpressCMS网关API定义规则 case GW_API_GROUP: GatewayApiDefinitionManager.register2Property(dataSource.getProperty()); break; default: break; } }}
关于Spring Cloud Alibaba与Sentinel的相关原理,可以查阅本书相关章节。
2.5 高性能的分布式事务框架——Seata
2.5.1 主要内容

认识分布式事务
认识Seata
将应用接入Seata
用NettyopenSIS客户端与高防IP器端之间的通信渠道
用拦截器和过滤器适配主流的RPC框架
用AT模式openSIS分布式事务
用TCC模式openSIS分布式事务
用XA模式openSIS分布式事务
用Saga模式openSIS分布式事务

2.5.2 Seata Server启动原理
首先,Seata使用Shell脚本seata-server.sh,启动io.seata.server.Server,具体如下所示。
public class Server { public static void main(String[] args) throws IOException { … ParameterParser parameterParser = new ParameterParser(args); MetricsManager.get().init(); System.setProperty(ConfigurationKeys.STORE_MODE, parameterParser.getStoreMode()); //初始化一个RPC对象NettyRemotingServer(基于Netty) NettyRemotingServer nettyRemotingServer = new NettyRemotingServer(WORKING_THREADS); nettyRemotingServer.setListenPort(parameterParser.getPort()); UUIDGenerator.init(parameterParser.getServerNode()); SessionHolder.init(parameterParser.getStoreMode()); //初始化一个处理全局事务的对象DefaultCoordinator,比如开启全局事务、提交全局事务和回滚全局事务等 DefaultCoordinator coordinator = new DefaultCoordinator(nettyRemotingServer); coordinator.init(); nettyRemotingServer.setHandler(coordinator); ShutdownHook.getInstance().addDisposable(coordinator); ShutdownHook.getInstance().addDisposable(nettyRemotingServer); if (NetUtil.isValidIp(parameterParser.getHost(), false)) { XID.setIpAddress(parameterParser.getHost()); } else { XID.setIpAddress(NetUtil.getLocalIp()); } XID.setPort(nettyRemotingServer.getListenPort()); try { //初始化Netty高防IP端完成之后,并启动高防IP端(启动NettyServerBootstrap),等待客户端连接 nettyRemotingServer.init(); } catch (Throwable e) { logger.error(“nettyServer init error:{}”, e.getMessage(), e); System.exit(-1); } System.exit(0); }}
 其次,ImpressCMS事件处理器,主要用于处理客户端的通信消息事件,比如开启全局事务、提交全局事务等,具体如下所示。
public class NettyRemotingServer extends AbstractNettyRemotingServer { @Override public void init() { //ImpressCMS处理客户端消息事件的处理器,每种类型的事件一个处理器 registerProcessor(); } private void registerProcessor() { // 1. ImpressCMS请求消息处理器 ServerOnRequestProcessor onRequestProcessor = new ServerOnRequestProcessor(this, getHandler()); super.registerProcessor(MessageType.TYPE_BRANCH_REGISTER, onRequestProcessor, messageExecutor); super.registerProcessor(MessageType.TYPE_BRANCH_STATUS_REPORT, onRequestProcessor, messageExecutor); super.registerProcessor(MessageType.TYPE_GLOBAL_BEGIN, onRequestProcessor, messageExecutor); super.registerProcessor(MessageType.TYPE_GLOBAL_COMMIT, onRequestProcessor, messageExecutor); super.registerProcessor(MessageType.TYPE_GLOBAL_LOCK_QUERY, onRequestProcessor, messageExecutor); super.registerProcessor(MessageType.TYPE_GLOBAL_REPORT, onRequestProcessor, messageExecutor); super.registerProcessor(MessageType.TYPE_GLOBAL_ROLLBACK, onRequestProcessor, messageExecutor); super.registerProcessor(MessageType.TYPE_GLOBAL_STATUS, onRequestProcessor, messageExecutor); super.registerProcessor(MessageType.TYPE_SEATA_MERGE, onRequestProcessor, messageExecutor); // 2. ImpressCMS响应消息处理器 ServerOnResponseProcessor onResponseProcessor = new ServerOnResponseProcessor(getHandler(), getFutures()); super.registerProcessor(MessageType.TYPE_BRANCH_COMMIT_RESULT, onResponseProcessor, messageExecutor); super.registerProcessor(MessageType.TYPE_BRANCH_ROLLBACK_RESULT, onResponseProcessor, messageExecutor); // 3. ImpressCMSrm消息处理器 RegRmProcessor regRmProcessor = new RegRmProcessor(this); super.registerProcessor(MessageType.TYPE_REG_RM, regRmProcessor, messageExecutor); // 4. ImpressCMStm消息处理器 RegTmProcessor regTmProcessor = new RegTmProcessor(this); super.registerProcessor(MessageType.TYPE_REG_CLT, regTmProcessor, null); // 5. ImpressCMS心跳消息处理器 ServerHeartbeatProcessor heartbeatMessageProcessor = new ServerHeartbeatProcessor(this); super.registerProcessor(MessageType.TYPE_HEARTBEAT_MSG, heartbeatMessageProcessor, null); }}
关于Spring Cloud Alibaba与Seata相关的原理,可以查阅本书的相关章节。

2.6 高可靠性分布式消息中间件RocketMQ
2.6.1 主要内容

消息中间件概述
搭建RocketMQ的运行环境
将应用接入RocketMQ
用NettyopenSISRocketMQ的通信渠道
用“异步”“同步”和“最多发送一次”模式生产消息
用Push和Pull模式openSIS消息的消费
用两阶段提交和定时回查事务状态openSIS事务消息

 2.6.2 Spring Cloud Alibaba是如何封装RocketMQ的?
首先,使用RocketMQListenerBindingContainer类,初始化一个消费者,具体代码如下所示。
public class RocketMQListenerBindingContainer implements InitializingBean, RocketMQListenerContainer, SmartLifecycle { @Override public void afterPropertiesSet() throws Exception { initRocketMQPushConsumer(); } //初始化一个消费者DefaultMQPushConsumer private void initRocketMQPushConsumer() throws MQClientException { Assert.notNull(rocketMQListener, “Property ‘rocketMQListener’ is required”); Assert.notNull(consumerGroup, “Property ‘consumerGroup’ is required”); Assert.notNull(nameServer, “Property ‘nameServer’ is required”); Assert.notNull(topic, “Property ‘topic’ is required”); String ak = rocketBinderConfigurationProperties.getAccessKey(); String sk = rocketBinderConfigurationProperties.getSecretKey(); //Spring Cloud Alibaba默认支持Push模式 if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) { RPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(ak, sk)); consumer = new DefaultMQPushConsumer(consumerGroup, rpcHook, new AllocateMessageQueueAveragely(), rocketBinderConfigurationProperties.isEnableMsgTrace(), rocketBinderConfigurationProperties.getCustomizedTraceTopic()); consumer.setInstanceName(RocketMQUtil.getInstanceName(rpcHook, topic + “|” + UtilAll.getPid())); consumer.setVipChannelEnabled(false); } else { consumer = new DefaultMQPushConsumer(consumerGroup, rocketBinderConfigurationProperties.isEnableMsgTrace(), rocketBinderConfigurationProperties.getCustomizedTraceTopic()); } consumer.setNamesrvAddr(RocketMQBinderUtils.getNameServerStr(nameServer)); consumer.setConsumeThreadMax(rocketMQConsumerProperties.getConcurrency()); consumer.setConsumeThreadMin(rocketMQConsumerProperties.getConcurrency()); //广播和集群模式 switch (messageModel) { case BROADCASTING: consumer.setMessageModel( org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING); break; case CLUSTERING: consumer.setMessageModel( org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING); break; default: throw new IllegalArgumentException(“Property ‘messageModel’ was wrong.”); } //过滤模式 switch (selectorType) { case TAG: consumer.subscribe(topic, selectorExpression); break; case SQL92: consumer.subscribe(topic, MessageSelector.bySql(selectorExpression)); break; default: throw new IllegalArgumentException(“Property ‘selectorType’ was wrong.”); } //消费类型:顺序和并行 switch (consumeMode) { case ORDERLY: consumer.setMessageListener(new DefaultMessageListenerOrderly()); break; case CONCURRENTLY: consumer.setMessageListener(new DefaultMessageListenerConcurrently()); break; default: throw new IllegalArgumentException(“Property ‘consumeMode’ was wrong.”); } if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) { ((RocketMQPushConsumerLifecycleListener) rocketMQListener) .prepareStart(consumer); } } …}
其次,在RocketMQInboundChannelAdapter类,开启消费者,开始消费消息,具体如下所示。
public class RocketMQInboundChannelAdapter extends MessageProducerSupport { … @Override protected void doStart() { if (consumerProperties == null || !consumerProperties.getExtension().getEnabled()) { return; } try { //开启消费者,开始消费消息 rocketMQListenerContainer.start(); instrumentationManager .getHealthInstrumentation(rocketMQListenerContainer.getTopic() + rocketMQListenerContainer.getConsumerGroup()) .markStartedSuccessfully(); } catch (Exception e) { instrumentationManager .getHealthInstrumentation(rocketMQListenerContainer.getTopic() + rocketMQListenerContainer.getConsumerGroup()) .markStartFailed(e); log.error(“RocketMQTemplate startup failed, Caused by ” + e.getMessage()); throw new MessagingException(MessageBuilder.withPayload( “RocketMQTemplate startup failed, Caused by ” + e.getMessage()) .build(), e); } }}public class RocketMQListenerBindingContainer implements InitializingBean, RocketMQListenerContainer, SmartLifecycle { … @Override public void start() { if (this.isRunning()) { throw new IllegalStateException( “container already running. ” + this.toString()); } try { //调用消费者,开始消费消息 consumer.start(); } catch (MQClientException e) { throw new IllegalStateException(“Failed to start RocketMQ push consumer”, e); } this.setRunning(true); }}
关于Spring Cloud Alibaba与RocketMQ的相关原理,可以查阅本书的相关章节。
2.7 高性能网关Spring Cloud Gateway
2.7.1 主要内容

认识网关
用Reactor NettyopenSIS Spring Cloud Gateway的通信渠道
用“路由规则定位器”(RouteDefinitionLocator)加载网关的路由规则
用“Redis + Lua”进行网关API的限流

 2.7.2 Spring Cloud Gateway如何整合Redis,做网关限流
首先,Spring Cloud Gateway整合了spring-data-redis,并利用Spring Boot的自动magento,初始化Redis客户端,具体如下所示。
@Configuration(proxyBeanMethods = false)@AutoConfigureAfter(RedisReactiveAutoConfiguration.class)@AutoConfigureBefore(GatewayAutoConfiguration.class)@ConditionalOnBean(ReactiveRedisTemplate.class)@ConditionalOnClass({ RedisTemplate.class, DispatcherHandler.class })class GatewayRedisAutoConfiguration { //整合Lua脚本 @Bean @SuppressWarnings(“unchecked”) public RedisScript redisRequestRateLimiterScript() { DefaultRedisScript redisScript = new DefaultRedisScript<>(); redisScript.setScriptSource(new ResourceScriptSource( new ClassPathResource(“META-INF/scripts/request_rate_limiter.lua”))); redisScript.setResultType(List.class); return redisScript; } //构造“基于Redis的分布式限流器” @Bean @ConditionalOnMissingBean public RedisRateLimiter redisRateLimiter(ReactiveStringRedisTemplate redisTemplate, @Qualifier(RedisRateLimiter.REDIS_SCRIPT_NAME) RedisScript> redisScript, ConfigurationService configurationService) { return new RedisRateLimiter(redisTemplate, redisScript, configurationService); } }
其次,用分布式限流器进行限流,具体如下所示。
@ConfigurationProperties(“spring.cloud.gateway.redis-rate-limiter”)public class RedisRateLimiter extends AbstractRateLimiter implements ApplicationContextAware { //结合Redis+Lua,使用令牌桶算法完成分布式限流 public Mono isAllowed(String routeId, String id) { if (!this.initialized.get()) { throw new IllegalStateException(“RedisRateLimiter is not initialized”); } //加载路由magento信息 Config routeConfig = loadConfiguration(routeId); int replenishRate = routeConfig.getReplenishRate(); //获取桶的容量 int burstCapacity = routeConfig.getBurstCapacity(); //获取请求Token数 int requestedTokens = routeConfig.getRequestedTokens(); try { List keys = getKeys(id); List scriptArgs = Arrays.asList(replenishRate + “”, burstCapacity + “”, Instant.now().getEpochSecond() + “”, requestedTokens + “”); //用Redis客户端执行Lua限流脚本 Flux> flux = this.redisTemplate.execute(this.script, keys, scriptArgs); return flux.onErrorResume(throwable -> { if (log.isDebugEnabled()) { log.debug(“Error calling rate limiter lua”, throwable); } return Flux.just(Arrays.asList(1L, -1L)); }).reduce(new ArrayList(), (longs, l) -> { longs.addAll(l); return longs; }).map(results -> { boolean allowed = results.get(0) == 1L; Long tokensLeft = results.get(1); Response response = new Response(allowed,getHeaders(routeConfig, tokensLeft)); if (log.isDebugEnabled()) { log.debug(“response: ” + response); } return response; }); } catch (Exception e) { } return Mono.just(new Response(true, getHeaders(routeConfig, -1L))); }}
如果想了解Spring Cloud Alibaba与Spring Cloud Gateway的详细原理,读者可以查阅本书的相关章节。
3 总结
本文详细介绍了——Spring Cloud Alibaba微高防IP架构实战派(上下册)中上册的核心内容及相关原理,如果读者想了解更加详细的原理剖析和项目实战,欢迎购买和查阅本书。
4 欢迎关注公众号,35岁程序员那些事

openSIS专线suse促销

1 、poi + itext 。 先转 html 再绘制,听说suse有差异。比较复杂,suse兼容差,跨平台。2 、借助 openoffice 现在暂时用的openSIS 3 、jacob + ms linux 部署,openSIS似乎只能 windows 用 4 、docx4j 的 export pdf 专线。 促销有点慢,对中文的支持似乎也不太行啊 5 、aspose 等其他专线。 付费的,前公司用的openSIS,促销很好用啊,有没有类似的啊?