SpringCloud集成Kafka

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();
        }
    }
}

猜你喜欢