首页 文章 精选 留言 我的

精选列表

搜索[文档编辑]，共10000篇文章
优秀的个人博客,低调大师

ApiBoot DataSource Switch 使用文档

ApiBoot是一款基于SpringBoot1.x，2.x的接口服务集成基础框架， 内部提供了框架的封装集成、使用扩展、自动化完成配置，让接口开发者可以选着性完成开箱即用， 不再为搭建接口框架而犯愁，从而极大的提高开发效率。 ApiBoot DataSource Switch 顾名思义，DataSource Switch是用于数据源选择切换的框架，这是一款基于Spring AOP切面指定注解实现的，通过简单的数据源注解配置就可以完成访问时的自动切换，DataSource Switch切换过程中是线程安全的。 添加依赖 使用DataSource Switch很简单，在pom.xml配置文件内添加如下依赖： <!--ApiBoot DataSource Switch--> <dependency> <groupId>org.minbox.framework</groupId> <artifactId>api-boot-starter-datasource-switch</artifactId> </dependency> ApiBoot所提供的依赖都不需要添加版本号，具体查看ApiBoot版本依赖 集成数据源实现 目前ApiBoot DataSource Switch集成了Druid、HikariCP两种数据源实现依赖，在使用方面也有一定的差异，因为每一个数据源的内置参数不一致。 Druid：参数配置前缀为api.boot.datasource.druid HikariCP：参数配置前缀为api.boot.datasource.hikari 具体使用请查看下面功能配置介绍。 配置参数 参数名 参数默认值 是否必填 参数描述 api.boot.datasource.primary master 否 主数据源名称 api.boot.datasource.druid.{poolName}.url 无 是 数据库连接字符串 api.boot.datasource.druid.{poolName}.username 无 是 用户名 api.boot.datasource.druid.{poolName}.password 无 是 密码 api.boot.datasource.druid.{poolName}.driver-class-name com.mysql.cj.jdbc.Driver 否 驱动类型 api.boot.datasource.druid.{poolName}.filters stat,wall,slf4j 否 Druid过滤 api.boot.datasource.druid.{poolName}.max-active 20 否 最大连接数 api.boot.datasource.druid.{poolName}.initial-size 1 否 初始化连接数 api.boot.datasource.druid.{poolName}.max-wait 60000 否 最大等待市场，单位：毫秒 api.boot.datasource.druid.{poolName}.validation-query select 1 from dual 否 检查sql api.boot.datasource.druid.{poolName}.test-while-idle true 否 建议配置为true，不影响性能，并且保证安全性。申请连接的时候检测，如果空闲时间大于timeBetweenEvictionRunsMillis，执行validationQuery检测连接是否有效。 api.boot.datasource.druid.{poolName}.test-on-borrow false 否 申请连接时执行validationQuery检测连接是否有效，做了这个配置会降低性能。 api.boot.datasource.druid.{poolName}.test-on-return false 否 归还连接时执行validationQuery检测连接是否有效，做了这个配置会降低性能。 api.boot.datasource.hikari.{poolName}.url 无 是 数据库连接字符串 api.boot.datasource.hikari.{poolName}.username 无 是 用户名 api.boot.datasource.hikari.{poolName}.password 无 是 密码 api.boot.datasource.hikari.{poolName}.driver-class-name com.mysql.cj.jdbc.Driver 否 数据库驱动类全限定名 api.boot.datasource.hikari.{poolName}.property 无 否 HikariCP属性配置 HikariCP数据源是SpringBoot2.x自带的，配置参数请访问HikariCP。 单主配置 ApiBoot DataSource Switch支持单主数据源的配置，application.yml配置文件如下所示： api: boot: datasource: # 配置使用hikari数据源 hikari: # master datasource config master: url: jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=Asia/Shanghai username: root password: 123456 修改主数据源名称 master为默认的主数据源的poolName，这里可以进行修改为其他值，不过需要对应修改primary参数，如下所示： api: boot: datasource: # 主数据源，默认值为master primary: main # 配置使用hikari数据源 hikari: # main datasource config main: url: jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=Asia/Shanghai username: root password: 123456 在上面配置主数据源的poolName修改为main。 主从配置 如果你的项目内存在单主单从、一主多从的配置方式，如下所示： api: boot: datasource: # 配置使用hikari数据源 hikari: # master datasource config master: url: jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=Asia/Shanghai username: root password: 123456 # 默认值为【com.mysql.cj.jdbc.Driver】 #driver-class-name: com.mysql.cj.jdbc.Driver # slave 1 datasource config slave_1: url: jdbc:mysql://localhost:3306/oauth2?characterEncoding=utf8&serverTimezone=Asia/Shanghai username: root password: 123456 # slave 2 datasource config slave_2: url: jdbc:mysql://localhost:3306/resources?characterEncoding=utf8&serverTimezone=Asia/Shanghai username: root password: 123456 在上面是一主多从的配置方式，分别是master、slave_1、slave_2。 多类型数据库配置 ApiBoot DataSource Switch提供了一个项目内连接多个不同类型的数据库，如：MySQL、Oracle...等，如下所示： api: boot: # 主数据源，默认值为master primary: mysql hikari: mysql: url: jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=Asia/Shanghai username: root password: 123456 oracle: url: jdbc:oracle:thin:@172.16.10.25:1521:torcl username: root password: 123456 driver-class-name: oracle.jdbc.driver.OracleDriver 在上面配置中，master主数据源使用的MySQL驱动连接MySQL数据库，而slave从数据源则是使用的Oracle驱动连接的Oracle数据库。 动态创建数据源 ApiBoot DataSource Switch内部提供了动态创建数据源的方法，可以通过注入ApiBootDataSourceFactoryBean来进行添加，如下所示： @Autowired private ApiBootDataSourceFactoryBean factoryBean; public void createNewDataSource() throws Exception { // 创建Hikari数据源 // 如果创建Druid数据源，使用DataSourceDruidConfig DataSourceHikariConfig config = new DataSourceHikariConfig(); // 数据库连接：必填 config.setUrl("jdbc:mysql://localhost:3306/resources"); // 用户名：必填 config.setUsername("root"); // 密码：必填 config.setPassword("123456"); // 数据源名称：必填(用于@DataSourceSwitch注解value值使用) config.setPoolName("dynamic"); // 创建数据源 DataSource dataSource = factoryBean.newDataSource(config); Connection connection = dataSource.getConnection(); System.out.println(connection.getCatalog()); connection.close(); } 自动切换 ApiBoot DataSource Switch的数据源自动切换主要归功于Spring的AOP，通过切面@DataSourceSwitch注解，获取注解配置的value值进行设置当前线程所用的数据源名称，从而通过AbstractRoutingDataSource进行数据源的路由切换。 我们沿用上面一主多从的配置进行代码演示，配置文件application.yml参考上面配置，代码示例如下： 从数据源示例类 @Service @DataSourceSwitch("slave") public class SlaveDataSourceSampleService { /** * DataSource Instance */ @Autowired private DataSource dataSource; /** * 演示输出数据源的catalog * * @throws Exception */ public void print() throws Exception { // 获取链接 Connection connection = dataSource.getConnection(); // 输出catalog System.out.println(this.getClass().getSimpleName() + " ->" + connection.getCatalog()); // 关闭链接 connection.close(); } } 主数据源示例类 @Service @DataSourceSwitch("master") public class MasterDataSourceSampleService { /** * DataSource Instance */ @Autowired private DataSource dataSource; /** * Slave Sample Service */ @Autowired private SlaveDataSourceSampleService slaveDataSourceSampleService; /** * 演示输出主数据源catalog * 调用从数据源类演示输出catalog * * @throws Exception */ public void print() throws Exception { Connection connection = dataSource.getConnection(); System.out.println(this.getClass().getSimpleName() + " ->" + connection.getCatalog()); connection.close(); slaveDataSourceSampleService.print(); } } 在主数据源的示例类内，我们通过@DataSourceSwitch("master")注解的value进行定位连接master数据源数据库。 同样在从数据库的示例类内，我们也可以通过@DataSourceSwitch("slave")注解的value进行定位连接slave数据源数据库。 单元测试示例 在上面的测试示例中，我们使用交叉的方式进行验证数据源路由是否可以正确的进行切换，可以编写一个单元测试进行验证结果，如下所示： @Autowired private MasterDataSourceSampleService masterDataSourceSampleService; @Test public void contextLoads() throws Exception { masterDataSourceSampleService.print(); } 运行上面测试方法，结果如下所示： MasterDataSourceSampleService ->test 2019-04-04 10:20:45.407 INFO 7295 --- [ main] com.zaxxer.hikari.HikariDataSource : HikariPool-2 - Starting... 2019-04-04 10:20:45.411 INFO 7295 --- [ main] com.zaxxer.hikari.HikariDataSource : HikariPool-2 - Start completed. SlaveDataSourceSampleService ->oauth2 单次执行数据源切换没有任何的问题，master数据源获取catalog输出后，调用slave示例类进行输出catalog。 ApiBoot DataSource Switch会在项目启动时首先初始化master节点DataSource实例，其他实例会在第一次调用时进行初始化。 压力性能测试 单次执行单线程操作没有问题，不代表多线程下不会出现问题，在开头说到过ApiBoot DataSource Switch是线程安全的，所以接下来我们来验证这一点，我们需要添加压力测试的依赖，如下所示： <dependency> <groupId>org.databene</groupId> <artifactId>contiperf</artifactId> <version>2.3.4</version> <scope>test</scope> </dependency> 接下来把上面的单元测试代码改造下，如下所示： // 初始化压力性能测试对象 @Rule public ContiPerfRule i = new ContiPerfRule(); @Autowired private MasterDataSourceSampleService masterDataSourceSampleService; /** * 开启500个线程执行10000次 */ @Test @PerfTest(invocations = 10000, threads = 500) public void contextLoads() throws Exception { masterDataSourceSampleService.print(); } 测试环境： 硬件：i7、16G、256SSD 系统：OS X 整个过程大约是10秒左右，ApiBoot DataSource Switch并没有发生出现切换错乱的情况。 注意事项 在使用ApiBoot DataSource Switch时需要添加对应数据库的依赖 如果使用Druid连接池，不要配置使用druid-starter的依赖，请使用druid依赖。 配置poolName时不要添加特殊字符、中文、中横线等。
2019-04-03
优秀的个人博客,低调大师

ApiBoot - ApiBoot Quartz 使用文档

ApiBoot Quartz ApiBoot内部集成了Quartz，提供了数据库方式、内存方式的进行任务的存储，其中数据库方式提供了分布式集群任务调度，任务自动平滑切换执行节点。 引用ApiBoot Quartz 在pom.xml配置文件内添加，如下配置： <!--ApiBoot Quartz--> <dependency> <groupId>org.minbox.framework</groupId> <artifactId>api-boot-starter-quartz</artifactId> </dependency> 备注：如果使用ApiBoot Quartz的内存方式，仅需要添加上面的依赖即可。 相关配置 参数名称 是否必填 默认值 描述 api.boot.quartz.job-store-type 否 memory 任务存储源方式，默认内存方式 api.boot.quartz.scheduler-name 否 scheduler 调度器名称 api.boot.quartz.auto-startup 否 true 初始化后是否自动启动调度程序 api.boot.quartz.startup-delay 否 0 初始化完成后启动调度程序的延迟。 api.boot.quartz.wait-for-jobs-to-complete-on-shutdown 否 false 是否等待正在运行的作业在关闭时完成。 api.boot.quartz.overwrite-existing-jobs 否 false 配置的作业是否应覆盖现有的作业定义。 api.boot.quartz.properties 否 Quartz自定义的配置属性，具体参考quartz配置 api.boot.quartz.jdbc 否 配置数据库方式的Jdbc相关配置 内存方式 ApiBoot Quartz在使用内存方式存储任务时，不需要做配置调整。 数据库方式 需要在application.yml配置文件内修改api.boot.quartz.job-store-type参数，如下所示： api: boot: quartz: # Jdbc方式 job-store-type: jdbc Quartz所需表结构 Quartz的数据库方式内部通过DataSource获取数据库连接对象来进行操作数据，所操作数据表的表结构是固定的，ApiBoot把Quartz所支持的所有表结构都进行了整理，访问Quartz支持数据库建表语句列表查看，复制执行对应数据库语句即可。 创建任务类 我们只需要让新建类集成QuartzJobBean就可以完成创建一个任务类，如下简单示例： /** * 任务定义示例 * 与Quartz使用方法一致，ApiBoot只是在原生基础上进行扩展，不影响原生使用 * <p> * 继承QuartzJobBean抽象类后会在项目启动时会自动加入Spring IOC * * @author：恒宇少年 - 于起宇 * <p> * DateTime：2019-03-28 17:26 * Blog：http://blog.yuqiyu.com * WebSite：http://www.jianshu.com/u/092df3f77bca * Gitee：https://gitee.com/hengboy * GitHub：https://github.com/hengboy */ public class DemoJob extends QuartzJobBean { /** * logger instance */ static Logger logger = LoggerFactory.getLogger(DemoJob.class); @Override protected void executeInternal(JobExecutionContext context) throws JobExecutionException { logger.info("定时任务Job Key ： {}", context.getJobDetail().getKey()); logger.info("定时任务执行时所携带的参数：{}", JSON.toJSONString(context.getJobDetail().getJobDataMap())); //...处理逻辑 } } 任务参数 在任务执行时传递参数是必须的，ApiBoot Quartz提供了比较方便的传递方式，不过最终Quartz会把传递的值都会转换为String类型数据。 任务Key默认值 ApiBoot Quartz的newJob方法所创建的定时任务，如果在不传递Job Key参数时，会默认使用UUID随机字符串作为Job Key以及Trigger Key。 自定义任务开始时间 任务开始时间可以通过startAtTime方法进行设置，在不设置的情况下，任务创建完成后会立刻执行。 Cron 表达式任务 创建Cron类型任务如下所示： String jobKey = apiBootQuartzService.newJob(ApiBootCronJobWrapper.Context() .jobClass(DemoJob.class) .cron("0/5 * * * * ?") .param( ApiBootJobParamWrapper.wrapper().put("param", "测试")) .wrapper()); Cron 表达式任务由ApiBootCronJobWrapper类进行构建。 上面的DemoJob任务类将会每隔5秒执行一次。 Loop 重复任务 Loop循环任务，当在不传递重复执行次数时，不进行重复执行，仅仅执行一次，如下所示： String jobKey = apiBootQuartzService.newJob( ApiBootLoopJobWrapper.Context() // 参数 .param( ApiBootJobParamWrapper.wrapper() .put("userName", "恒宇少年") .put("userAge", 24) ) // 每次循环的间隔时间，单位：毫秒 .loopIntervalTime(2000) // 循环次数 .repeatTimes(5) // 开始时间，10秒后执行 .startAtTime(new Date(System.currentTimeMillis() + 10000)) // 任务类 .jobClass(DemoJob.class) .wrapper() ); Loop 任务由ApiBootLoopJobWrapper类进行构建。 上面的定时任务将会重复执行5次，连上自身执行的一次也就是会执行6次，每次的间隔时间为2秒，在任务创建10秒后进行执行。 Once 一次性任务 Once一次性任务，任务执行一次会就会被自动释放，如下所示： Map paramMap = new HashMap(1); paramMap.put("paramKey", "参数值"); String jobKey = apiBootQuartzService.newJob( ApiBootOnceJobWrapper.Context() .jobClass(DemoJob.class) // 参数 .param( ApiBootJobParamWrapper.wrapper() .put("mapJson", JSON.toJSONString(paramMap)) ) // 开始时间，2秒后执行 .startAtTime(new Date(System.currentTimeMillis() + 2000)) .wrapper() ); Once 任务由ApiBootOnceJobWrapper类进行构建。 在参数传递时可以是对象、集合，不过需要进行转换成字符串才可以进行使用。 暂停任务执行 任务在执行过程中可以进行暂停操作，通过ApiBoot Quartz提供的pauseJob方法就可以很简单的实现，当然暂停时需要传递Job Key，Job Key可以从创建任务方法返回值获得。 暂停任务如下所示： // 暂停指定Job Key的任务 apiBootQuartzService.pauseJob(jobKey); // 暂停多个执行中任务 apiBootQuartzService.pauseJobs(jobKey,jobKey,jobKey); 恢复任务执行 任务执行完暂停后，如果想要恢复可以使用如下方式： // 恢复指定Job Key的任务执行 apiBootQuartzService.resumeJob(jobKey); // 恢复多个暂停任务 apiBootQuartzService.resumeJobs(jobKey,jobKey,jobKey); 修改Cron表达式 修改Cron表达式的场景如下： 已创建 & 未执行 已创建 & 已执行 修改方法如下所示： // 修改执行Job Key任务的Cron表达式 apiBootQuartzService.updateJobCron(jobKey, "0/5 * * * * ?"); 删除任务 想要手动释放任务时可以使用如下方式： // 手动删除指定Job Key任务 apiBootQuartzService.deleteJob(jobKey); // 手动删除多个任务 apiBootQuartzService.deleteJobs(jobKey,jobKey,jobKey); 删除任务的顺序如下： 暂停触发器 移除触发器 删除任务本章源码地址：https://github.com/hengboy/api-boot/tree/master/api-boot-samples/api-boot-sample-quartz
2019-04-03
优秀的个人博客,低调大师

Apache Storm 官方文档 —— Metrics

Storm 提供了一个可以获取整个拓扑中所有的统计信息的度量接口。Storm 内部通过该接口可以跟踪各类统计数字：executor 和 acker 的数量、每个 bolt 的平均处理时延、worker 使用的最大堆容量等等，这些信息都可以在 Nimbus 的 UI 界面中看到。 Metric 类型 使用 Metrics 只需要实现一个接口方法：getValueAndReset，在方法中可以查找汇总值、并将该值复位为初始值。例如，在 MeanReducer 中就实现了通过运行总数除以对应的运行计数的方式来求取均值，然后将两个值都重新设置为 0。 Storm 提供了以下几种 metric 类型： AssignableMetric— 将 metric 设置为指定值。此类型在两种情况下有用：1. metric 本身为外部设置的值；2. 你已经另外计算出了汇总的统计值。 CombinedMetric— 可以对 metric 进行关联更新的通用接口。 CountMetric— 返回 metric 的汇总结果。可以调用incr()方法来将结果加一；调用incrBy(n)方法来将结果加上给定值。 MultiCountMetric— 返回包含一组 CountMetric 的 HashMap ReducedMetric MeanReducer— 跟踪由它的reduce()方法提供的运行状态均值结果（可以接受Double、Integer、Long等类型，内置的均值结果是Double类型）。MeanReducer 确实是一个相当棒的家伙。 MultiReducedMetric— 返回包含一组 ReducedMetric 的 HashMap Metric Consumer 构建自定义 metric 内建的 Metric builtin_metrics.clj为内建的 metrics 设置了数据结构，以及其他框架组件可以用于更新的虚拟方法。metrics 本身是在回调代码中实现计算的 — 请参考executor.clj中的ack-spout-msg的例子。 转载自并发编程网 - ifeve.com
2017-05-21
优秀的个人博客,低调大师

Apache Storm 官方文档 —— Hooks

Storm 提供了一种 hooks 机制，可以实现在 Storm 的各种事件流中运行自定义代码的功能。可以通过继承BaseTaskHook类来创建 hook，还可以根据需要在继承的子类中覆写适当的方法来跟踪相关事件。 注册 hook 有两种方法： 在 spout 的 open 方法或者 bolt 的 prepare 方法中使用TopologyContext#addTaskHook方法； 使用 Storm 配置表中的topology.auto.task.hooks配置项。之后这些 hook 会自动注册到每个 spout 和 bolt 中，这样就可以很方便地处理例如集成自定义的系统监控代码之类的事情了。 转载自并发编程网 - ifeve.com
2017-05-21
优秀的个人博客,低调大师

Apache Storm 官方文档 —— FAQ

Storm 最佳实践 关于配置 Storm + Trident 的建议 worker 的数量最好是服务器数量的倍数；topology 的总并发度(parallelism)最好是 worker 数量的倍数；Kafka 的分区数(partitions)最好是 Spout（特指KafkaSpout）并发度的倍数 在每个机器（supervisor）上每个拓扑应用只配置一个 worker 在拓扑最开始运行的时候设置使用较少的大聚合器，并且最好是每个 worker 进程分配一个 使用独立的调度器（scheduler）来分配任务（关于Scheduler 的知识请参考xumingming 的博客—— 译者注） 在每个 worker 上只配置使用一个 acker —— 这是 0.9.x 版本的默认特性，不过在早期版本中有所不同 在配置文件中开启 GC 日志记录；如果一切正常，日志中记录的 major GC 应该会非常少 将 trident 的 batch interval 配置为你的集群的端到端时延的 50% 左右 开始时设置一个很小的TOPOLOGY_MAX_SPOUT_PENDING（对于 trident 可以设置为 1，对于一般的 topology 可以设置为 executor 的数量），然后逐渐增大，直到数据流不再发生变化。这时你可能会发现结果大约等于“2 × 吞吐率(每秒收到的消息数) × 端到端时延”（最小的额定容量的2倍）。 如何避免 worker 总是出现莫名其妙的故障的问题 确保 Storm 对你的日志目录有写权限 确保你的堆内存没有溢出 确保所有的 worker 上都已经正确地安装了所有的依赖库文件 确保 ZooKeeper 的 hostname 不是简单地设置为 “localhost” 确保集群中的每台机器上都配置好了正确、唯一的 hostname，并且这些 hostname 需要配置在所有机器的 Storm 配置文件中 确保 a) 不同的 worker 之间，b) 不同的 Storm 节点之间，c) Storm 与 ZooKeeper 集群之间， d) 各个 worker 与拓扑运行所需要的 Kafka/Kestrel/其他数据库等 之间没有开启防火墙或者其他安全保护机制；如果有，请使用 netstat 来为各个端口之间的通信授权 Help！Storm 使用过程中无法获取： 日志文件：日志文件默认记录在$STORM_HOME/logs目录中。请检查你对该目录是否有写权限。具体的日志配置信息位于 logback/cluster.xml 文件中（0.9 之前的版本需要在 log4j/*.properties 配置文件中进行配置。 最终输出的 JVM 设置：需要在配置文件（storm.yaml）的childopts配置项中添加-XX+PrintFlagsFinal命令选项。 最终输出的 Java 系统属性信息：需要在你构建拓扑的位置添加代码Properties props = System.getProperties(); props.list(System.out); 我应该使用多少个 worker？ worker 的完整数量是由 supervisor 配置的。每个 supervisor 会分配到一定数量的 JVM slot，你在拓扑中设置的 worker number 就是以这个 slot 数量为依据进行分配的。 不建议为每个拓扑在每台机器上分配超过一个 worker。 假如有一个运行于三台 8 核服务器节点的拓扑，它的并行度为24，每个 bolt 在每台机器上分配有 8 个 executor（即每个 CPU 核心分配一个）。这种场景下，使用三个 worker （每个 worker 分配 8 个executor）相对于使用更多的 worker （比如使用 24 个 worker，为每个 executor 分别分配一个）有三点好处： 首先，在 worker 内部将数据流重新分区到不同的 executor 的操作（比如 shuffle 或者 group-by）就不会产生触发到传输 buffer 缓冲区，tuple 会直接从发送端转储到接收端的 buffer 区。这一点有很大的优势。相反，如果目标 executor 是在同一台机器的不同 worker 进程内，tuple 就需要经历“发送 -> worker 传输队列 -> 本地 socket 端口 -> 接收端 worker -> 接收端 executor”这样一个漫长的过程。虽然这个过程并不会产生网络级传输，但是在同一台机器的不同进程间的传输损耗也是很可观的。 其次，三个大的聚合器带来的大的缓存空间比 24 个小聚合器带来的小缓存空间要有用得多。因为这回降低数据倾斜造成的影响，同时提高 LRU 的性能。 最后，更少的 worker 可以有效地降低控制流的频繁变动。 拓扑 Trident 拓扑支持多数据流吗 Trident 拓扑可以设计成条件路径（if-else）的工作流形式吗？例如，bolt0 在接收 spout 的数据流时，可以根据输入 tuple 的值来选择将数据流发送到 bolt1 或者 bolt2，而不是同时向两个 bolt 发送。 Trident 的 “each” 运算符可以返回一个数据流对象，你可以将该对象存储在某个变量中，然后你可以对同一个数据流执行多个 each 操作来分解该数据流，如下述代码所示： Stream s = topology.each(...).groupBy(...).aggregate(...) Stream branch1 = s.each(..., FilterA) Stream branch2 = s.each(..., FilterB) 你可以使用 join、merge 或者 multiReduce 来联结各个数据流。 到目前为止，Trident 暂时不支持输出多个数据流。（详见STORM-68） Spout Coordinator 是什么，为什么会有很多 Coordinator？ Trident spout 实际上是通过 Storm 的 bolt 运行的。MasterBatchCoordinator（MBC）封装了 Trident 拓扑的 spout，它负责整合 Trident 中的 batch，这一点对于你所使用的任何类型的 spout 而言都是一样的。Trident 的 batch 就是在 MBC 向各个 spout-coordinator 分发种子 tuple 的过程中生成的。Spout-coordinator bolt 知道你所定义的 spout 是如何互相协作的 —— 实际上，在使用 Kafka 的情况下，各个 spout 就是通过 spout-coordinator 来获取 pull 消息所需要的 partition 和 offset 信息的。 在 spout 的 metadata 记录中能够存储什么信息？ 只能存储少量静态数据，而且是越少越好（尽管你确实可以向其中存储更多的信息，不过我们不推荐这样做）。 emitPartitionBatchNew函数是多久调用一次的？ 由于在 Trident 中 MBC 才是实际运行的 spout，一个 batch 中的所有 tuple 都是 MBC 生成的 tuple 树的节点。也就是说，Storm 的 “max spout pending” 参数实际上定义的是可以并发运行的 batch 数量。MBC 在满足以下两个条件下会发送出一个新的 batch：首先，挂起的 tuple 数需要小于 “max pending” 参数；其次，距离上一个 batch 的发送已经过去了至少一个trident batch interval的间隔时间。 如果没有数据发送，Trident 会降低发送频率吗？ 是的，Storm 中有一个可选的 “spout 等待策略”，默认配置是 sleep 一段指定的配置时间。 Trident batch interval 参数有什么用？ 你知道 486 时代的计算机上面为什么有个trubo button吗？这个参数的作用和这个按钮有点像。 实际上，trident batch interval 有两个用处。首先，它可以用于减缓 spout 从远程数据源获取数据的速度，但这不会影响数据处理的效率。例如，对于一个从给定的 S3 存储区中读取批量上传文件并按行发送数据的 spout，我们就不希望它经常触发 S3 的阈值，因为文件要隔几分钟才会上传一次，而且每个 batch 也需要花费一定的时间来执行。 另一个用处是限制启动期间或者突发数据负载情况下内部消息队列的负载压力。如果 spout 突然活跃起来，并向系统中挤入了 10 个 batch 的记录，那么可能会有从 batch7 开始的大量不紧急的 tuple 堵塞住传输缓冲区，并且阻塞了从 batch3 中的 tuple（甚至可能包含 batch3 中的部分旧 tuple）的 commit 过程#。对于这种情况，我们的解决方法就是将 trident batch interval 设置为正常的端到端处理时延的一半左右 —— 也就是说如果需要花费 600 ms 的时间处理一个 batch，那么就可以每 300 ms 处理一个 batch。 注意，这个 300 ms 仅仅是一个上限值，而不是额外增加的延时时间，如果你的 batch 需要花费 258 ms 来运行，那么 Trident 就只会延时等待 42 ms。 如何设置 batch 大小？ Trident 本身不会对 batch 进行限制。不过如果使用 Kafka 的相关 spout，那么就可以使用 max fetch bytes 大小除以 平均 record 大小来计算每个子 batch 分区的有效 record 大小。 怎样重新设置 batch 的大小？ Trident 的 batch 在某种意义上是一种过载的设施。batch 大小与 partition 的数量均受限于或者是可以用于定义#： 事务安全单元（一段时间内存在风险的 tuple）； 相对于每个 partition，一个用于窗口数据流分析的有效窗口机制； 相对于每个 partition，使用 partitionQuery，partitionPersist 等命令时能够同时进行的查询操作数量； 相对于每个 partition，spout 能够同时分配的 record 数量。 不能在 batch 生成之后更改 batch 的大小，不过可以通过 shuffle 操作以及修改并行度的方式来改变 partition 的数量。 时间相关问题 怎样基于指定时间聚合数据 对于带有固定时间戳的 records，如果需要对他们执行计数、求均值或者聚合操作，并将结果整合到离散的时间桶（time bucket）中，Trident 是一个很好的具有可扩展性的解决方案。 这种情况下可以写一个each函数来将时间戳置入一个时间桶中：如果桶的大小是以“小时”为单位的，那么时间戳2013-08-08 12:34:56就会被匹配到2013-08-08 12:00:00桶中，其他的 12 时到 13 时之间的时间也一样。然后可以使用persistentAggregate来对时间桶分组。persistentAggregate会使用一个基于数据存储的本地 cacheMap。这些包含有大量 records 的 group 会使用高效的批量读取/写入方式对数据存储区进行操作，所以并不会对数据存储区进行大量的读操作；只要你的数据传送足够快捷，Trident 就可以高效地使用内存与网络。即使某台服务器宕机了一天，需要重新快速地发送一整天的数据，旧有的结果也可以静默地获取到并进行更新，并且这并不会影响当前结果的计算过程。 怎么才能知道某个时间桶中已经收到了所有需要的 record？ 很遗憾，你不会知道什么时候所有的 event 都已经采集到了 —— 这是一个认识论问题，而不是一个分布式系统的问题。你可以： 使用域相关知识来设定时间限制。 引入标记机制：对于一个指定时间窗，确定某个 record 会处在所有的 record 的最后位置。Trident 使用这个机制来判断一个 batch 是否结束。例如，你收到一组传感器采集到的 records，每个传感器都是按顺序发送数据的，那么一旦所有的传感器都发送出一个 “3:02:xx” 的数据，你就可以知道可以开始处理这个时间窗了。 如果可以的话，尽量使你的处理过程增量化：每个新来的值都会使结果越来越准确。Trident ReducerAggregator 就是一个可以通过一个旧有的结果以及一组新数据来返回一个更新的结果的运算符。这使得结果可以被缓存并序列化到一个数据库中；如果某台服务器宕机了一天，在恢复运行之后需要重新快速地发送一整天的数据，旧有的结果也可以静默地获取到并进行更新。 使用 Lambda 架构：将所有收到的事件数据归档到存储区中（S3，HBase，HDFS）。在快速处理层，一旦时间窗复位，就对对应的时间桶进行处理来获取有效结果，并且在处理过程中跳过所有比早于该时间窗的过期数据。定期地执行全局聚合操作就可以计算出一个较“正确”的结果。 附注 #此处译文可能不够准确，有疑问的读者请参考原文对应内容。 转载自并发编程网 - ifeve.com
2017-05-21

资源下载

更多资源
腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

时间: 2026-08-21 大小: 1MB 下载: 10次
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

用户登录
用户注册