依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
配置
spring:
kafka:
bootstrap-servers: 外网ip:9092
producer:
retries: 0
batch-size: 16384
buffer-memory: 33554432
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
group-id: test
auto-offset-reset: earliest
enable-auto-commit: true
auto-commit-interval: 100
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
生产者
消息实体类
@AllArgsConstructor
@Data
@NoArgsConstructor
@ToString
public class Message {
private Long id;
private String msg;
private Date time;
}
@Component
public class KafkaProducer {
private static final String TOPIC = "pktest";
@Autowired
private KafkaTemplate kafkaTemplate;
@SuppressWarnings("unchecked")
@Scheduled(fixedRate = 1000 * 10)
public void produce() {
Message message = new Message(12L,"helloword",new Date());
kafkaTemplate.send(TOPIC, JSON.toJSONString(message));
}
}
消费者
@Component
@Slf4j
public class KafkaConsumer {
@KafkaListener(topics = "pktest")
public void consumer(String message) {
log.info("收到的消息:" + message);
}
}