亚游,  基于对行业的长期关注和此次简单的下海调研(跟蓝领工友们吃炸鸡、喝啤酒、住宿舍),针对昆山制造业蓝领(最具代表性),以下从他们的吃穿、住房、交友、借贷、培训等方向,把最真实的情况展现给创业者和投资人,并给出了我对这个切入点以及商业模式的一些看法。

国家行政学院公共管理教研部教授竹立家告诉《法制日报》记者:“延迟退休是大政策,各工种、各性别的差异都会考虑到的,肯定不会‘一刀切’。”女人没经验开什么店好  2014年8月7日,中国对WU-14助推滑翔高超音速滑翔飞行器(HGV)进行了第二次测试。此次从山西省太原卫星发射中心实施发射的测试实际上没有达到预期落点。

发布于 2018-03-23 15:58:13 | 131 次阅读 | 评论: 0 | 来源: 网友投递

这里有新鲜出炉的精品教程,程序狗速度看过来!

Spring Boot

Spring Boot 项目旨在简化创建产品级的 Spring 应用和服务。你可通过它来选择不同的 Spring 平台。可创建独立的 Java 应用和 Web 应用,同时提供了命令行工具来允许 'spring scripts'.


本篇文章主要介绍了在springboot中对kafka进行读写的示例代码,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧

springboot对kafka的client很好的实现了集成,使用非常方便,本文也实现了一个在springboot中实现操作kafka的demo。

1.POM配置

只需要在dependencies中增加 spring-kafka的配置即可。完整效果如下:


<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>1.5.4.RELEASE</version>
  </parent>

  <properties>
    <java.version>1.8</java.version>
     <spring-kafka.version>1.2.2.RELEASE</spring-kafka.version>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
   <!-- spring-kafka -->
      <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
      <version>${spring-kafka.version}</version>
      </dependency>
      <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka-test</artifactId>
      <version>${spring-kafka.version}</version>
      <scope>test</scope>
      </dependency>
   </dependencies>

2.生产者

参数配置类,其参数卸载yml文件中,通过@Value注入


package com.dhb.kafka.producer;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class SenderConfig {

  @Value("${kafka.bootstrap-servers}")
  private String bootstrapServers;

  @Bean
  public Map<String,Object> producerConfigs() {
    Map<String,Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,this.bootstrapServers);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
    props.put(ProducerConfig.ACKS_CONFIG,"0");
    return props;
  }

  @Bean
  public ProducerFactory<String,String> producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs());
  }

  @Bean
  public KafkaTemplate<String,String> kafkaTemplate() {
    return new KafkaTemplate<String, String>(producerFactory());
  }

  @Bean
  public Sender sender() {
    return new Sender();
  }
}

消息发送类


package com.dhb.kafka.producer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

@Slf4j
public class Sender {

  @Autowired
  private KafkaTemplate<String,String> kafkaTemplate;

  public void send(String topic,String payload) {
    log.info("sending payload='{}' to topic='{}'",payload,topic);
    this.kafkaTemplate.send(topic,payload);
  }
}

3.消费者

参数配置类


package com.dhb.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class ReceiverConfig {

  @Value("${kafka.bootstrap-servers}")
  private String bootstrapServers;

  public Map<String,Object> consumerConfigs() {
    Map<String,Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG,"helloword");
    return props;
  }

  @Bean
  public ConsumerFactory<String,String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
  }

  @Bean
  public ConcurrentKafkaListenerContainerFactory<String,String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String,String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;
  }

  @Bean
  public Receiver receiver() {
    return new Receiver();
  }

}

消息接受类


package com.dhb.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;

import java.util.concurrent.CountDownLatch;

@Slf4j
public class Receiver {

  private CountDownLatch latch = new CountDownLatch(1);

  public CountDownLatch getLatch() {
    return latch;
  }

  @KafkaListener(topics = "${kafka.topic.helloworld}")
  public void receive(String payload) {
    log.info("received payload='{}'",payload);
    latch.countDown();
  }
}

3.web测试类

定义了一个基于http的web测试接口


package com.dhb.kafka.web;

import com.dhb.kafka.producer.Sender;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;

@RestController
@Slf4j
public class KafkaProducer {

  @Autowired
  Sender sender;

  @RequestMapping(value = "/sender.action", method = RequestMethod.POST)
  public void exec(HttpServletRequest request, HttpServletResponse response,String data) throws IOException{
    this.sender.send("testtopic",data);
    response.setCharacterEncoding("UTF-8");
    response.setContentType("text/json");
    response.getWriter().write("success");
    response.getWriter().flush();
    response.getWriter().close();
  }

}

4.启动类及配置


package com.dhb.kafka;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class KafkaApplication {


  public static void main(String[] args) {
    SpringApplication.run(KafkaApplication.class,args);

  }
}

application.yml


kafka:
 bootstrap-servers: 192.168.162.239:9092
 topic:
  helloworld: testtopic

程序结构:

包结构

5.读写测试

通过执行KafkaApplication的main方法启动程序。然后打开postman进行测试:


运行后返回success


生产者日志:


消费者日志:


以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持PHPERZ。



相关阅读 :
亚游在springboot中对kafka进行读写的示例代码
亚游在springboot中对kafka进行读写的示例代码
亚游在springboot中对kafka进行读写的示例代码
springboot前后台数据交互的示例代码
springboot+maven快速构建项目的示例代码
springboot+maven快速构建项目的示例代码
springboot+maven快速构建项目的示例代码
SpringBoot配置发送Email的示例代码
SpringBoot配置发送Email的示例代码
SpringBoot配置发送Email的示例代码
详解SpringBoot中实现依赖注入功能
详解SpringBoot缓存的实例代码(EhCache 2.x 篇)
最新网友评论  共有(0)条评论 发布评论 返回顶部

Copyright © 2007-2017 亚游 All Rights Reserved   冀ICP备14009818号  亚游  广告服务
美国议会候选人发誓若当选就用区块链记录选民反馈 山东主管疫苗疾控官员突然住院 自杀传闻无法证实 韩朝举行将军级会谈 讨论《板门店宣言》履行方案 两高今年要干这些大事 将影响你的生活 科技部长:欢迎各国电动汽车都到中国市场来 刘宇婕又赢了!别克中国青少年精英赛夺E组冠军 法院通报朋友圈侵权案:微信不是法外之地 言论应规制 轰炸叙境内伊军事目标后 以总理再对伊叙强硬警告 外媒:英新外交大臣首访华 中英承诺加强自贸合作 西蒙尼:托雷斯去不去中超?这你得去问他! 中国空军发布歼-16宣传片 飞行画面披露(图) 这些食物吃一口胖十斤 想瘦的人千万别碰
这位副科长为获重用 冒充中央部委下指示当晚被抓 拜仁大将:C罗比梅西更全面 看看欧冠的进球数 德美正渐行渐远? 美媒:德企反特朗普 德民众反美 叙利亚总统:白头盔成员将被视作恐怖分子消灭掉 台渔船被日舰艇水炮威胁 台当局被质疑装聋作哑 为这件事 19名博士后挂职北京延庆区进行一年锻炼 竞彩大势:门兴继续连胜 波尔图小胜可期 韩国总统特使将访俄 通报访朝访美成果 半岛局势和缓台海变成新\"火药桶\"?国台办回应 商务部谈中美经贸磋商:已做好准备应对各种可能 羽毛球世锦赛8个常识 林丹5枚金牌史上并列第一 数据趣谈:勇士比肩80年代绿军 甜瓜得分超名宿
三星研发Exynos 5G调制解调器 为年新机做准备 欧文缺阵绿军遭逆转 比尔34分奇才双加时过关 英媒:中非关系越来越紧密 最多贷款流向这个领域 安倍与瑞典国王会面 确认合作使日欧经济协定生效 取关母队巨星又一骚操作! 不给顶薪哥就走? 5G标准之争 联想亲历者讲述“投票”细节 山东高铁通道进出口仅3个 路网密度居全国第12位 贵人资本梁渊:料港股近期走势反复 可留意绩优股 土耳其东南部安全形势仍严峻 外交部吁暂勿前往 两万元投资 小投资冷门暴利行业 没学历女生做什么工作 需要钱的的给我打电话 亚游