一、疑问描述
spring-kafka通过 @KafkaListener 的方式配置订阅的topic,通过@Configuration 配置创建kafkaListenerContainerFactory。
如下:
@Configuration
@EnableKafka
public class KafkaConfig {
private static final String KAFKA_SERVERS_CONFIG = "10.192.77.202:9092";
private static final String LOCAL_GROUP_ID = "test";
@Bean
ConcurrentKafkaListenerContainerFactory<Integer, String>
kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
@Bean
public ConsumerFactory<Integer, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS_CONFIG);
props.put(ConsumerConfig.GROUP_ID_CONFIG, LOCAL_GROUP_ID);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return props;
}
@Bean
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS_CONFIG);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<String, String>(producerFactory());
}
@Bean
public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@KafkaListener(topics = "TEST_TOPIC_NEW")
public void listen(String data) {
System.out.println("kafkaconfig =listen======="+data);
}
}
但想要动态的创建监听者对象,如通过数据库的方式配置KAFKA_SERVERS_CONFIG 和LOCAL_GROUP_ID ,并且可以不用重启服务,实现热更新。通过spring-kafka提供的接口没有找到好的解决方法。
二、解决方案
所以,考虑通过最基本的手动创建消费者对象。
通过定时任务,每三分钟check一次,从数据库读取相应配置,将已有配置写入缓存,当读取的配置和缓存不一致时,销毁已有消费者,创建新的消费者。
如果有好的方案,谢谢告知~
/**
* 每三分钟check一次kafka配置
* @throws Exception
*/
@Scheduled(cron = "1 1/3 * * * ? ")
public void deviceNotifyConfig(){
Map<String, String> kafkaConfigs = systemConfigService.fetchConfigLikeKey("kafka");
if(kafkaConfigs != null && kafkaConfigs.size() != 0)
{
String kafkaIp = kafkaConfigs.get("kafkaIp");
String kafkaPort = kafkaConfigs.get("kafkaPort");
String kafkaUserName = kafkaConfigs.get("kafkaUserName");
String kafkaPassword = kafkaConfigs.get("kafkaPassword");
if(StringUtils.isNotEmpty(KafkaLinkCache.kafkaConfigCache))
{
if (!KafkaLinkCache.kafkaConfigCache.equals(kafkaIp + "_" + kafkaPort))
{
//关闭已有消费者对象
KafkaConsumer<String, String> consumer = KafkaLinkCache.DEVICE_CONSUMER_MAP.get("kafkaComsumer");
if(consumer != null)
{
resourceNotifyConsumer.closeConsumer();
}
KafkaLinkCache.DEVICE_CONSUMER_MAP.clear();
this.handlerConsumer(kafkaIp, kafkaPort);
}
}
else
{
this.handlerConsumer(kafkaIp, kafkaPort);
}
}else
{
//关闭已有消费者对象
KafkaConsumer<String, String> consumer = KafkaLinkCache.DEVICE_CONSUMER_MAP.get("kafkaComsumer");
if(consumer != null)
{
resourceNotifyConsumer.closeConsumer();
}
KafkaLinkCache.DEVICE_CONSUMER_MAP.clear();
}
}
private void handlerConsumer(String kafkaIp, String kafkaPort) {
Properties props = new Properties();
props.setProperty("bootstrap.servers", kafkaIp + ":" + kafkaPort);
// key反序列化
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// value反序列化
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 每个消费者都必须属于某一个消费组,所以必须指定group.id
props.put("group.id", "test");
// 构造消费者对象
deviceNoifyThreadExecutor.execute(()->{
KafkaConsumer<String, String> consumerObj = null;
// 指定多主题:
List<String> topics = CbdmOptUtil.stringToStringList(PropertiesUtil.getProperty("kafka.subscribe.topics"), ConstParamErrorCode.DEFAULT_SPLIT_KEY, false);
try {
consumerObj = new KafkaConsumer<>(props);
if(consumerObj != null) {
consumerObj.subscribe(topics);
resourceNotifyConsumer.setConsumer(consumerObj);
KafkaLinkCache.DEVICE_CONSUMER_MAP.put("kafkaComsumer", consumerObj);
resourceNotifyConsumer.onMessage();
}
} catch(Exception e) {
LogUtils.logError(RunTimeLogUtil.toErrorLog(ConstParamErrorCode.SYSTEM_CODE_FAIL + "", LogObjectTypeEnum.SYSTEM,"consume",
"resolve data platform notify error"),e);
}finally {
// 关闭
consumerObj.close();
}
});
//保存配置
KafkaLinkCache.kafkaConfigCache = kafkaIp + "_" + kafkaPort;
}
@Component(value = "resourceNotifyConsumer")
public class ResourceNotifyConsumer {
private Logger logger = LoggerFactory.getLogger(ResourceNotifyConsumer.class);
@Resource
IAccessDeviceService resourceService;
private KafkaConsumer<String, String> consumer = null;
public KafkaConsumer<String, String> getConsumer() {
return consumer;
}
public void setConsumer(KafkaConsumer<String, String> consumer) {
this.consumer = consumer;
}
public void closeConsumer()
{
//consumer非线程安全,依靠gc回收
consumer = null;
}
public void onMessage(){
try{
logger.info(RunTimeLogUtil.toLog(LogObjectTypeEnum.SYSTEM,"consume","Get resource Notify start",null,null));
while (true) {
if(consumer != null)
{
// timeout 阻塞时间,从kafka中取出100毫秒的数据,有可能一次取出0到N条
List<Map<String,Object>> datas = new ArrayList<>();
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
// 遍历
for (ConsumerRecord<String, String> record : records) {
Map<String,Object> notifyDto = ( Map<String,Object> ) JsonUtils.jsonToMap(record.value());
datas.add(notifyDto);
}
// 拿出结果
if(CollectionUtils.isNotEmpty(datas)){
logger.info(RunTimeLogUtil.toLog(LogObjectTypeEnum.SYSTEM,"consume","Get resource Notify",null,null, "record"),JsonUtils.object2Json(datas));
// 起线程处理 资源变更通知
resourceHandle(datas);
}
} else {
break;
}
}
}catch (Throwable e){
logger.error(RunTimeLogUtil.toErrorLog(ConstParamErrorCode.SYSTEM_CODE_FAIL + "",LogObjectTypeEnum.SYSTEM,"consume",
"resolve resource notify error"),e);
}
}
/**
*
* @param datas
*/
private void resourceHandle(List<Map<String,Object>> datas){
if(CollectionUtils.isNotEmpty(datas)){
try {
new Thread(() -> resourceService.dealResource(datas)).start();
}catch (Throwable e){
logger.error(RunTimeLogUtil.toErrorLog(ConstParamErrorCode.SYSTEM_CODE_FAIL + "",LogObjectTypeEnum.SYSTEM,"consume",
"resourceHandle error"),e);
}
}else{
logger.info(RunTimeLogUtil.toLog(LogObjectTypeEnum.SYSTEM,"consume","resource notify data is empty!",null,null));
}
}
}
文章浏览阅读1.8k次,点赞4次,收藏6次。python简易爬虫v1.0作者:William Ma (the_CoderWM)进阶python的首秀,大部分童鞋肯定是做个简单的爬虫吧,众所周知,爬虫需要各种各样的第三方库,例如scrapy, bs4, requests, urllib3等等。此处,我们先从最简单的爬虫开始。首先,我们需要安装两个第三方库:requests和bs4。在cmd中输入以下代码:pip install requestspip install bs4等安装成功后,就可以进入pycharm来写爬虫了。爬
文章浏览阅读2.6k次。解决方法:解决方法可以去github重新下载一个pyflakes.vim。执行如下命令git clone --recursive git://github.com/kevinw/pyflakes-vim.git然后进入git克降目录,./pyflakes-vim/ftplugin,通过如下命令将python目录下的所有文件复制到~/.vim/ftplugin目录下即可。cp -R ...._freetorn.vim
文章浏览阅读210次,点赞7次,收藏3次。本文简述了hello.c源程序的预处理、编译、汇编、链接和运行的主要过程,以及hello程序的进程管理、存储管理与I/O管理,通过hello.c这一程序周期的描述,对程序的编译、加载、运行有了初步的了解。_hit csapp
文章浏览阅读1w次,点赞2次,收藏27次。来源:机器人小妹 很多时候企业拥有重复,乏味且困难的工作流程,这些流程往往会减慢生产速度并增加运营成本。为了降低生产成本,企业别无选择,只能自动化某些功能以降低生产成本。 通过数字化..._人工智能平台
文章浏览阅读2.2k次。热加载能够在每次保存修改的代码后自动刷新 electron 应用界面,而不必每次去手动操作重新运行,这极大的提升了开发效率。安装 electron 热加载插件热加载虽然很方便,但是不是每个 electron 项目必须的,所以想要舒服的开发 electron 就只能给 electron 项目单独的安装热加载插件[electron-reloader]:// 在项目的根目录下安装 electron-reloader,国内建议使用 cnpm 代替 npmnpm install electron-relo._electron-reloader
文章浏览阅读942次。在11.0 进行定制化开发,会根据需要去掉recovery模式的一些选项 就是在device.cpp去掉一些选项就可以了。_android recovery 删除 部分菜单
文章浏览阅读2.2k次,点赞2次,收藏6次。继续上次的echart博客,由于省会流向图是从echart画廊中直接取来的。所以直接上代码<!DOCTYPE html><html><head> <meta charset="utf-8" /> <meta name="viewport" content="width=device-width,initial-scale=1,minimum-scale=1,maximum-scale=1,user-scalable=no" /&_java+echart地图+物流跟踪
文章浏览阅读1.4k次。一、OSD模块简介1.1 消息封装:在OSD上发送和接收信息。cluster_messenger -与其它OSDs和monitors沟通client_messenger -与客户端沟通1.2 消息调度:Dispatcher类,主要负责消息分类1.3 工作队列:1.3.1 OpWQ: 处理ops(从客户端)和sub ops(从其他的OSD)。运行在op_tp线程池。1...._ceph 发送数据到其他副本的源码
文章浏览阅读7.9k次,点赞3次,收藏22次。一 定义这是最早出现的置换算法。该算法总是淘汰最先进入内存的页面,即选择在内存中驻留时间最久的页面予以淘汰。该算法实现简单,只需把一个进程已调入内存的页面,按先后次序链接成一个队列,并设置一个指针,称为替换指针,使它总是指向最老的页面。但该算法与进程实际运行的规律不相适应,因为在进程中,有些页面经常被访问,比如,含有全局变量、常用函数、例程等的页面,FIFO 算法并不能保证这些页面不被淘汰。这里,我_进程调度fifo算法代码
文章浏览阅读133次。rownum是oracle才有的写法,rownum在oracle中可以用于取第一条数据,或者批量写数据时限定批量写的数量等mysql取第一条数据写法SELECT * FROM t order by id LIMIT 1;oracle取第一条数据写法SELECT * FROM t where rownum =1 order by id;ok,上面是mysql和oracle取第一条数据的写法对比,不过..._mysql 替换@rownum的写法
文章浏览阅读790次,点赞3次,收藏4次。官网下载下载链接:http://www.eclipse.org/downloads/点击Download下载完成后双击运行我选择第2个,看自己需要(我选择企业级应用,如果只是单纯学习java选第一个就行)进入下一步后选择jre和安装路径修改jvm/jre的时候也可以选择本地的(点后面的文件夹进去),但是我们没有11版本的,所以还是用他的吧选择接受安装中安装过程中如果有其他界面弹出就点accept就行..._ecjelm
文章浏览阅读245次。原文链接:https://linux.cn/article-7801-1.htmlifconfigping <IP地址>:发送ICMP echo消息到某个主机traceroute <IP地址>:用于跟踪IP包的路由路由:netstat -r: 打印路由表route add :添加静态路由路径routed:控制动态路由的BSD守护程序。运行RIP路由协议gat..._ifconfig 删除vlan