解释：
Kafka是一个分布式的消息存储系统,提供了四大核心接口：
1.Producer API允许了应用可以向Kafka中的topics发布消息；
2.Consumer API允许了应用可以订阅Kafka中的topics,并消费消息；
3.Streams API允许应用可以作为消息流的处理者，比如可以从topicA中消费消息，处理的结果发布到topicB中；
4.Connector API提供Kafka与现有的应用或系统适配功能，比如与数据库连接器可以捕获表结构的变化；
Topic —> 每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic.
Producer —> 负责发布消息到Kafka broker.
Consumer —> 消息消费者，向Kafka broker读取消息的客户端.
Kafka安装:
Kafka下载地址：(http://kafka.apache.org/downloads)
目录
Windows启动方式：
分别启动Zookeeper、Kafka
\bin\windows\zookeeper-server-start.bat config\zookeeper.properties
\bin\windows\kafka-server-start.bat config\server.properties
提供kafka服务不需要在本地安装。
Spring Boot整合Kafaka：
非注解使用方式：
pom引入：
<!--kafka-clients发送消息所需jar包-->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>1.0.0</version>
</dependency>
添加配置.properties文件：
#============== kafka ===================
# 指定kafka 代理地址，可以多个
spring.kafka.bootstrap-servers=localhost:9092
#=============== provider =======================
spring.kafka.producer.retries=0 设置大于0的值，则客户端会将发送失败的记录重新发送
# 每次批量发送消息的数量
spring.kafka.producer.batch-size=16384
spring.kafka.producer.buffer-memory=33554432
# 指定消息key和消息体的编解码方式 UTF-8
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
编写一个生产者者:
package com.zhongway.modules.kafka.provider;
import com.google.common.io.Resources;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.zhongway.modules.kafka.entity.MessageEntity;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.InputStream;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.Future;
/**
* @author Minko
*/
public class KafkaSender {
private static Logger logger = LoggerFactory.getLogger(KafkaSender.class);
private static KafkaProducer<String, String> producer;
private Gson gson = new GsonBuilder().create();
static {
try {
InputStream props = Resources.getResource("producer.props").openStream();
Properties properties = new Properties();
properties.load(props);
producer = new KafkaProducer<>(properties);
} catch (IOException e) {
logger.error("初始化Kafka配置文件失败");
}
}
/**
* 发送消息方法
*
* @param topic 主题
* @param msg 消息体
*/
public void sendMsg(String topic, String msg) {
MessageEntity message = new MessageEntity();
message.setMsg(msg);
message.setSendTime(new Date());
logger.info("sendMessage = {}", gson.toJson(message));
try {
Future<RecordMetadata> record = producer.send(new ProducerRecord<>(topic, gson.toJson(message)));
record.get();
} catch (Exception e) {
logger.error("sendErrorMessage = {}", gson.toJson(message));
}
}
}
更简单的使用注解方式：
pom引入：
对应版本
此处引入2.1.x 其对应kafka-clients版本为所需的1.0.0
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.1.0.RELEASE</version>
</dependency>
添加配置文件application.yml:
kafka:
bootstrap-servers: localhost:9092 # 指定kafka 代理地址，可以多个
producer: # 生产者
retries: 1 # 设置大于0的值，则客户端会将发送失败的记录重新发送
# 每次批量发送消息的数量
batch-size: 16384
buffer-memory: 33554432
# 指定消息key和消息体的编解码方式
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
编写一个生产者:
/**
* 生产者
* @author Minko
*/
@Component
public class KafkaSender {
@Autowired
private KafkaTemplate<String,String> kafkaTemplate;
/**
* 发送消息到kafka
*@param topic 主题
*@param message 内容体
*/
public void sendMsg(String topic , String message){
kafkaTemplate.send(topic ,message);
}
}
两种方式外部只需提供 topic 和发送的 json字符串即可 。
关于定时任务，更新后的renren框架取消了job 中 method，每个定时任务需要实现ITask的 run方法，源码中会获取存入schedule_job表中的bean名称 和 run方法根据cron表达式去执行该方法。
示例：
@Component("KafkaSenderTask ")
public class KafkaSenderTask implements ITask {
private Logger logger = LoggerFactory.getLogger(getClass());
/**
* params 可为空
* @param params 参数，多参数使用JSON数据
*/
@Override
public void run(String params){
KafkaSender kafkaSender = new KafkaSender();
kafkaSender.sendMsg("M","工单消息内容");
}
}
kafka葵花宝典：传送门=>
Demo:
public class GsonTest {
public static void main(String[] args) {
List<Map<String, String>> mapList = new ArrayList<>();
Map map = new HashMap();
map.put("id", "1");
map.put("name", "葵花宝典");
Map map2 = new HashMap();
map2.put("id", "2");
map2.put("name", "九阴真经");
mapList.add(map);
mapList.add(map2);
Gson gson = new Gson();
System.out.println(gson.toJson(mapList));
}
}
打印结果：[{"name":"葵花宝典","id":"1"},{"name":"九阴真经","id":"2"}]
...end