首页 文章 精选 留言 我的

精选列表

搜索[实时语音],共10000篇文章
优秀的个人博客,低调大师

NSQ 1.2.0 发布,实时分布式消息传递平台

NSQ 是无中心设计、节点自动注册和发现的开源消息系统。可作为内部通讯框架的基础,易于配置和发布。NSQ 1.2.0 已经发布,如下: 升级 nsqd:删除了对 v0.3.8 及更早版本中使用的旧元数据方案的支持(你不能直接从 0.3.8 升级到 v1.2.0,而是通过 v1.0.0-compat 或 v1.1.0) 使用 Go 模块管理依赖项:dep 目前仍有支持,但不建议使用 特性 nsqd: 添加--max-channel-consumers 添加--min-output-buffer-timeout 限制使用者可以请求的超时时间(默认值 --max-output-buffer-timeout 提高到30秒) 添加标志以调整默认值--output-buffer-timeout 为 Authd 请求添加随机负载平衡 在 Authd 请求中包括客户端 TLS cert CommonName 添加参数到 /stats 端点以允许跳过每个客户端状态 将生产者客户端 tcp 连接添加到 stats nsq_to_file: 在大多数日志消息中包括topic/channel 添加--log-level和--log-prefix 大重构,更健壮的文件切换、同步和错误处理 支持使用 --work-dir 的单独工作目录 新标志 --sync-interval nsq_to_http:添加--header nsqadmin:添加--base-path 修复 nsqd: 关闭不发送“magic”标题的连接 nsqd、nsqlookupd、nsqadmin:重构日志级别,一般重构以更好地退出所有致命错误 nsqadmin:切换到像 nsqd 和 nsqadmin 那样使用 JudWhite/go-svc、修复客户端计数和信道总消息率、Elint 报告修复/清理 详情见更改日志: https://github.com/nsqio/nsq/releases/tag/v1.2.0

优秀的个人博客,低调大师

SpringBoot2.0 整合 QuartJob ,实现定时器实时管理

一、QuartJob简介 1、一句话描述 Quartz是一个完全由java编写的开源作业调度框架,形式简易,功能强大。 2、核心API (1)、Scheduler 代表一个 Quartz 的独立运行容器,Scheduler 将 Trigger 绑定到特定 JobDetail, 这样当 Trigger 触发时, 对应的 Job 就会被调度。 (2)、Trigger 描述 Job 执行的时间触发规则。主要有 SimpleTrigger 和 CronTrigger 两个子类,通过一个 TriggerKey 唯一标识。 (3)、Job 定义一个任务,规定了任务是执行时的行为。JobExecutionContext 提供了调度器的上下文信息,Job 的数据可从 JobDataMap 中获取。 (4)、JobDetail Quartz 在每次执行 Job 时,都重新创建一个 Job 实例,所以它不直接接受一个 Job 的实例,相反它接收一个 Job 实现类。描述 Job 的实现类及其它相关的静态信息,如 Job 名字、描述等。 二、与SpringBoot2.0 整合 1、项目结构 版本描述 spring-boot:2.1.3.RELEASE quart-job:2.3.0 2、定时器配置 import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.quartz.SchedulerFactoryBean; import javax.sql.DataSource; import java.util.Properties; @Configuration public class ScheduleConfig { @Bean public SchedulerFactoryBean schedulerFactoryBean(DataSource dataSource) { // Quartz参数配置 Properties prop = new Properties(); // Schedule调度器的实体名字 prop.put("org.quartz.scheduler.instanceName", "HuskyScheduler"); // 设置为AUTO时使用,默认的实现org.quartz.scheduler.SimpleInstanceGenerator是基于主机名称和时间戳生成。 prop.put("org.quartz.scheduler.instanceId", "AUTO"); // 线程池配置 prop.put("org.quartz.threadPool.class", "org.quartz.simpl.SimpleThreadPool"); prop.put("org.quartz.threadPool.threadCount", "20"); prop.put("org.quartz.threadPool.threadPriority", "5"); // JobStore配置:Scheduler在运行时用来存储相关的信息 // JDBCJobStore和JobStoreTX都使用关系数据库来存储Schedule相关的信息。 // JobStoreTX在每次执行任务后都使用commit或者rollback来提交更改。 prop.put("org.quartz.jobStore.class", "org.quartz.impl.jdbcjobstore.JobStoreTX"); // 集群配置:如果有多个调度器实体的话则必须设置为true prop.put("org.quartz.jobStore.isClustered", "true"); // 集群配置:检查集群下的其他调度器实体的时间间隔 prop.put("org.quartz.jobStore.clusterCheckinInterval", "15000"); // 设置一个频度(毫秒),用于实例报告给集群中的其他实例 prop.put("org.quartz.jobStore.maxMisfiresToHandleAtATime", "1"); // 触发器触发失败后再次触犯的时间间隔 prop.put("org.quartz.jobStore.misfireThreshold", "12000"); // 数据库表前缀 prop.put("org.quartz.jobStore.tablePrefix", "qrtz_"); // 从 LOCKS 表查询一行并对这行记录加锁的 SQL 语句 prop.put("org.quartz.jobStore.selectWithLockSQL", "SELECT * FROM {0}LOCKS UPDLOCK WHERE LOCK_NAME = ?"); // 定时器工厂配置 SchedulerFactoryBean factory = new SchedulerFactoryBean(); factory.setDataSource(dataSource); factory.setQuartzProperties(prop); factory.setSchedulerName("HuskyScheduler"); factory.setStartupDelay(30); factory.setApplicationContextSchedulerContextKey("applicationContextKey"); // 可选,QuartzScheduler 启动时更新己存在的Job factory.setOverwriteExistingJobs(true); // 设置自动启动,默认为true factory.setAutoStartup(true); return factory; } } 3、定时器管理工具 import com.quart.job.entity.ScheduleJobBean; import org.quartz.*; /** * 定时器工具类 */ public class ScheduleUtil { private ScheduleUtil (){} private static final String SCHEDULE_NAME = "HUSKY_" ; /** * 触发器 KEY */ public static TriggerKey getTriggerKey(Long jobId){ return TriggerKey.triggerKey(SCHEDULE_NAME+jobId) ; } /** * 定时器 Key */ public static JobKey getJobKey (Long jobId){ return JobKey.jobKey(SCHEDULE_NAME+jobId) ; } /** * 表达式触发器 */ public static CronTrigger getCronTrigger (Scheduler scheduler,Long jobId){ try { return (CronTrigger)scheduler.getTrigger(getTriggerKey(jobId)) ; } catch (SchedulerException e){ throw new RuntimeException("getCronTrigger Fail",e) ; } } /** * 创建定时器 */ public static void createJob (Scheduler scheduler, ScheduleJobBean scheduleJob){ try { // 构建定时器 JobDetail jobDetail = JobBuilder.newJob(TaskJobLog.class).withIdentity(getJobKey(scheduleJob.getJobId())).build() ; CronScheduleBuilder scheduleBuilder = CronScheduleBuilder .cronSchedule(scheduleJob.getCronExpression()) .withMisfireHandlingInstructionDoNothing() ; CronTrigger trigger = TriggerBuilder.newTrigger() .withIdentity(getTriggerKey(scheduleJob.getJobId())) .withSchedule(scheduleBuilder).build() ; jobDetail.getJobDataMap().put(ScheduleJobBean.JOB_PARAM_KEY,scheduleJob); scheduler.scheduleJob(jobDetail,trigger) ; // 如果该定时器处于暂停状态 if (scheduleJob.getStatus() == 1){ pauseJob(scheduler,scheduleJob.getJobId()) ; } } catch (SchedulerException e){ throw new RuntimeException("createJob Fail",e) ; } } /** * 更新定时任务 */ public static void updateJob(Scheduler scheduler, ScheduleJobBean scheduleJob) { try { // 构建定时器 TriggerKey triggerKey = getTriggerKey(scheduleJob.getJobId()); CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(scheduleJob.getCronExpression()) .withMisfireHandlingInstructionDoNothing(); CronTrigger trigger = getCronTrigger(scheduler, scheduleJob.getJobId()); trigger = trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(scheduleBuilder).build(); trigger.getJobDataMap().put(ScheduleJobBean.JOB_PARAM_KEY, scheduleJob); scheduler.rescheduleJob(triggerKey, trigger); // 如果该定时器处于暂停状态 if(scheduleJob.getStatus() == 1){ pauseJob(scheduler, scheduleJob.getJobId()); } } catch (SchedulerException e) { throw new RuntimeException("updateJob Fail",e) ; } } /** * 停止定时器 */ public static void pauseJob (Scheduler scheduler,Long jobId){ try { scheduler.pauseJob(getJobKey(jobId)); } catch (SchedulerException e){ throw new RuntimeException("pauseJob Fail",e) ; } } /** * 恢复定时器 */ public static void resumeJob (Scheduler scheduler,Long jobId){ try { scheduler.resumeJob(getJobKey(jobId)); } catch (SchedulerException e){ throw new RuntimeException("resumeJob Fail",e) ; } } /** * 删除定时器 */ public static void deleteJob (Scheduler scheduler,Long jobId){ try { scheduler.deleteJob(getJobKey(jobId)); } catch (SchedulerException e){ throw new RuntimeException("deleteJob Fail",e) ; } } /** * 执行定时器 */ public static void run (Scheduler scheduler, ScheduleJobBean scheduleJob){ try { JobDataMap dataMap = new JobDataMap() ; dataMap.put(ScheduleJobBean.JOB_PARAM_KEY,scheduleJob); scheduler.triggerJob(getJobKey(scheduleJob.getJobId()),dataMap); } catch (SchedulerException e){ throw new RuntimeException("run Fail",e) ; } } } 4、定时器执行和日志 import com.quart.job.entity.ScheduleJobBean; import com.quart.job.entity.ScheduleJobLogBean; import com.quart.job.service.ScheduleJobLogService; import org.quartz.JobExecutionContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.quartz.QuartzJobBean; import java.lang.reflect.Method; import java.util.Date; /** * 定时器执行日志记录 */ public class TaskJobLog extends QuartzJobBean { private static final Logger LOG = LoggerFactory.getLogger(TaskJobLog.class) ; @Override protected void executeInternal(JobExecutionContext context) { ScheduleJobBean jobBean = (ScheduleJobBean)context.getMergedJobDataMap().get(ScheduleJobBean.JOB_PARAM_KEY) ; ScheduleJobLogService scheduleJobLogService = (ScheduleJobLogService)SpringContextUtil.getBean("scheduleJobLogService") ; // 定时器日志记录 ScheduleJobLogBean logBean = new ScheduleJobLogBean () ; logBean.setJobId(jobBean.getJobId()); logBean.setBeanName(jobBean.getBeanName()); logBean.setParams(jobBean.getParams()); logBean.setCreateTime(new Date()); long beginTime = System.currentTimeMillis() ; try { // 加载并执行定时器的 run 方法 Object target = SpringContextUtil.getBean(jobBean.getBeanName()); Method method = target.getClass().getDeclaredMethod("run", String.class); method.invoke(target, jobBean.getParams()); long executeTime = System.currentTimeMillis() - beginTime; logBean.setTimes((int)executeTime); logBean.setStatus(0); LOG.info("定时器 === >> "+jobBean.getJobId()+"执行成功,耗时 === >> " + executeTime); } catch (Exception e){ // 异常信息 long executeTime = System.currentTimeMillis() - beginTime; logBean.setTimes((int)executeTime); logBean.setStatus(1); logBean.setError(e.getMessage()); } finally { scheduleJobLogService.insert(logBean) ; } } } 三、定时器服务封装 1、定时器初始化 @Service public class ScheduleJobServiceImpl implements ScheduleJobService { @Resource private Scheduler scheduler ; @Resource private ScheduleJobMapper scheduleJobMapper ; /** * 定时器初始化 */ @PostConstruct public void init (){ ScheduleJobExample example = new ScheduleJobExample() ; List<ScheduleJobBean> scheduleJobBeanList = scheduleJobMapper.selectByExample(example) ; for (ScheduleJobBean scheduleJobBean : scheduleJobBeanList) { CronTrigger cronTrigger = ScheduleUtil.getCronTrigger(scheduler,scheduleJobBean.getJobId()) ; if (cronTrigger == null){ ScheduleUtil.createJob(scheduler,scheduleJobBean); } else { ScheduleUtil.updateJob(scheduler,scheduleJobBean); } } } } 2、添加定时器 @Override @Transactional(rollbackFor = Exception.class) public int insert(ScheduleJobBean record) { ScheduleUtil.createJob(scheduler,record); return scheduleJobMapper.insert(record); } 3、立即执行一次定时器 @Override @Transactional(rollbackFor = Exception.class) public void run(Long jobId) { ScheduleJobBean scheduleJobBean = scheduleJobMapper.selectByPrimaryKey(jobId) ; ScheduleUtil.run(scheduler,scheduleJobBean); } 4、更新定时器 @Override @Transactional(rollbackFor = Exception.class) public int updateByPrimaryKeySelective(ScheduleJobBean record) { ScheduleUtil.updateJob(scheduler,record); return scheduleJobMapper.updateByPrimaryKeySelective(record); } 5、停止定时器 @Override @Transactional(rollbackFor = Exception.class) public void pauseJob(Long jobId) { ScheduleJobBean scheduleJobBean = scheduleJobMapper.selectByPrimaryKey(jobId) ; ScheduleUtil.pauseJob(scheduler,jobId); scheduleJobBean.setStatus(1); scheduleJobMapper.updateByPrimaryKeySelective(scheduleJobBean) ; } 6、恢复定时器 @Override @Transactional(rollbackFor = Exception.class) public void resumeJob(Long jobId) { ScheduleJobBean scheduleJobBean = scheduleJobMapper.selectByPrimaryKey(jobId) ; ScheduleUtil.resumeJob(scheduler,jobId); scheduleJobBean.setStatus(0); scheduleJobMapper.updateByPrimaryKeySelective(scheduleJobBean) ; } 7、删除定时器 @Override @Transactional(rollbackFor = Exception.class) public void delete(Long jobId) { ScheduleUtil.deleteJob(scheduler, jobId); scheduleJobMapper.deleteByPrimaryKey(jobId) ; } 四、配置一个测试的定时器 1、定时接口封装 public interface TaskService { void run(String params); } 2、测试定时器 @Component("getTimeTask") public class GetTimeTask implements TaskService { private static final Logger LOG = LoggerFactory.getLogger(GetTimeTask.class.getName()) ; private static final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ; @Override public void run(String params) { LOG.info("Params === >> " + params); LOG.info("当前时间::::"+format.format(new Date())); } } 五、源代码地址 GitHub地址:知了一笑 https://github.com/cicadasmile/middle-ware-parent 码云地址:知了一笑 https://gitee.com/cicadasmile/middle-ware-parent

优秀的个人博客,低调大师

基于Spark SQL实现对HDFS操作的实时监控报警

1.前言 E-MapReduce计划从EMR-3.18.1版本开始提供Spark Streaming SQL的预览版功能。Spark Streaming SQL是在Spark Structured Streaming的基础上做了进一步封装,方便用户使用SQL语言进行Spark流式分析开发。Spark Streaming SQL直接地透明地受惠于Spark SQL的优化带来的性能提升,同时也遵循Spark Structured Streaming的语法约束,例如Spark Structured Streaming不支持多流聚合查询,Spark Streaming SQL也就同样不支持。关于EMR Spark Streaming SQL的使用入门,请参考: EMR Spark Streaming SQL语法参考 Spark Streamin

优秀的个人博客,低调大师

HBase2.0重新定义小对象实时存取

小对象,特别指1K~10MB范围的数据,比如图片,短视频,文档等。这些数据广泛的存在于人工智能,医疗,教育,生活分享,电子商务等领域。目前对象存储典型技术方案为AWS的S3以及阿里云的OSS,还有一些基于MySQL+对象存储的二次开发方案。这些方案解决了对象存储的可靠性和扩展性问题,但是存在一些缺陷:两个方案都存在访问延时问题,因为访问数据至少要2次查询,一次索引访问+一次数据访问,特别的当用户想查询一组有关联数据的时候,需要N次调用;对象存储很难将对象和一些维度、指标数据混合存储,而这些是查询时过滤数据的条件,导致对象存储的方案在检索能力上不足;MySQL+对象存储方案将维度、指标数据存储在Mysql解决了检索能力,但是引入了数据不一致问题,实现主备双活也非常困难,造成可用性上的缺陷,同时这个方案对用户代码侵入大,对运维也不友好

优秀的个人博客,低调大师

G7在实时计算的探索与实践

作者: 张皓 G7业务快览 G7主要通过在货车上的传感器感知车辆的轨迹、油耗、点熄火、载重、温度等数据,将车辆、司机、车队、货主连接到一起,优化货物运输的时效、安全、成本等痛点问题。 整个数据是通过车载的传感器设备采集,比如公司的Smart盒子,CTBox盒子,油感设备,温度探头等,将车辆数据上报到后端平台,在后端平台计算和处理,最后展示到用户面前。 G7的业务场景是典型的IoT场景: 传感器数据 数据种类多 数据质量差 数据低延迟 数据量大 其中,数据质量差的原因是整个链条会非常的长,从传感器采集的车辆的数据,通过网络运营商将数据上报到后端服务器,再经过解析,mq,过滤,调用三方接口,业务处理,入库,整个过程非常的长,造成数据在传输过程中出现数据重复,数据缺失等。另外一点,IoT场景需要数据传输的延迟非常低,比如进出区域报警,当车辆进入到某个电子围

资源下载

更多资源
Mario

Mario

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

Spring

Spring

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

Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能,例如代码缩略图,Python的插件,代码段等。还可自定义键绑定,菜单和工具栏。Sublime Text 的主要功能包括:拼写检查,书签,完整的 Python API , Goto 功能,即时项目切换,多选择,多窗口等等。Sublime Text 是一个跨平台的编辑器,同时支持Windows、Linux、Mac OS X等操作系统。

WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源,继承了IntelliJ IDEA强大的JS部分的功能。

用户登录
用户注册