kafka和springboot以及如何使用事務

• 1、引入依賴

  <dependency>
              <groupId>org.springframework.kafka</groupId>
              <artifactId>spring-kafka</artifactId>
              <version>2.2.4.RELEASE</version>
          </dependency>
  //這個是kafka的
  
  

  <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>log4j-over-slf4j</artifactId>
  </dependency>
  //這個是slf4j的
  
  
  

• 2、配置文件：

  server:
    port: 9001
    servlet:
      context-path: /kafka
  spring: kafka: producer: bootstrap-servers: 10.203.103.66:9092         //生產者的
  

     consumer:
        bootstrap-servers: 10.203.103.66:9092     //這個是消費者的
  logging:
    level:
      root: INFO

  
  
  //紅色的是kafka的配置，藍色的是本地的地址

   

• 3、消息的發送和消息的監聽

  @RestController
  @RequestMapping("kafka")
  public class KafkaController {
  
      @Autowired
      private KafkaTemplate template;
      private static final String topic="heima";//設置主題topic
      private static final Logger LOGGER=LoggerFactory.getLogger(KafkaApplication.class);
      @RequestMapping("index")
      public String index(){
          return "hello,kafka";
      }
  
  
      /*
      *
      * 消息的生產者
      * */
  
      @GetMapping("/send/{input}")
      public String sendToKafka(@PathVariable String input){
          this.template.send(topic,input);
          return "send success!"+input;
      }
  //這里懶得可以保證把input傳進來
      /*
      * 消息的接收
      * */
      @KafkaListener(id="myContainer1",topics = topic,groupId = "groud.demo-2")
      public void listener(String input){
         LOGGER.info(input);
      }

  如何使用事務控制回滾（已經發了一個消息，但是后面的消息出錯了，然后全部回滾）

1. 事務的支持：

   spring:
     kafka:
       producer:
         bootstrap-servers: 10.203.103.66:9092
         transaction-id-prefix: kafka_tx.
       consumer:
         bootstrap-servers: 10.203.103.66:9092
   
   //標紅的是事務的支持

    

2. 代碼部分：控制事務的方式

   //編碼方式
    @GetMapping("/send/{input}")
       public String sendToKafka(@PathVariable String input){
   //        this.template.send(topic,input);
           //事務的支持
           template.executeInTransaction(t->{
               t.send(topic,input);
               if("error".equals(input))
               {
                   throw new RuntimeException("input is error");
               }
               t.send(topic,input+"anthor");
               return true;
           });
   
           return "send success!"+input;
       }
   
   //第二種  （注解方式）
       @GetMapping("/send2/{input}")
       @Transactional(rollbackFor = RuntimeException.class)
       public String sendToKafka2(@PathVariable String input){
   //        this.template.send(topic,input);
           //事務的支持
           
               template.send(topic,input);
               if("error".equals(input))
               {
                   throw new RuntimeException("input is error");
               }
               template.send(topic,input+"anthor");
              
                return "send success!"+input;
              
           }