SpringCloud整合Kafka
阿新 • • 發佈:2019-02-02
1:maven依賴
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>Camden.SR6</version >
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId >spring-cloud-stream-binder-kafka</artifactId>
</dependency>
2:provider (.yml檔案)
server: port: 8081 spring: application: name: lilili-provider cloud: instance-count: 1 instance-index: 0 stream: kafka: binder: brokers: localhost:9092 zk-nodes: localhost:2182 auto-add-partitions: true auto-create-topics: true min-partition-count: 1 bindings: output: destination: mytopic content-type: text/plain producer: partitionCount: 1
@EnableBinding(Source.class)
public class SendService {
@Autowired
private Source source;
public void sendMessage(String msg) {
try {
source.output().send(MessageBuilder.withPayload(msg).build());
} catch (Exception e) {
e.printStackTrace();
}
}
}
@RestController
public class ProducerController {
@Autowired
private SendService service;
@RequestMapping(value = "/send/{msg}", method = RequestMethod.GET)
public void send(@PathVariable("msg") String msg){
service.sendMessage(msg);
}
}
3:consumer (.yml檔案)
spring:
application:
name: input-demo
cloud:
instance-count: 1
instance-index: 0
stream:
kafka:
binder:
brokers: localhost:9092
zk-nodes: localhost:2182
auto-add-partitions: true
auto-create-topics: true
min-partition-count: 1
bindings:
input:
destination: event-demo
group: s1
consumer:
autoCommitOffset: false
concurrency: 1
partitioned: false
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
/**
* lilili 2018/4/14.
*/
@EnableBinding(Sink.class)
public class KafkaConsumer {
@StreamListener(Sink.INPUT)
public void onReceive(byte[] msg) {
System.out.println("get Kafka message:" + new String(msg));
}
}
或者:
@EnableBinding(Sink.class)
public class MsgSink {
@StreamListener(Sink.INPUT)
public void process(Message<?> message) {
System.out.println(message.getPayload());
Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
if (acknowledgment != null) {
System.out.println("Acknowledgment provided");
acknowledgment.acknowledge();
}
}
}