pom配置

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>springcloud-demo</artifactId>
  7.         <groupId>demo</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  12.     <artifactId>demo-stream-rabbit-provider</artifactId>
  14.     <dependencies>
  15.         <dependency>
  16.             <groupId>org.springframework.boot</groupId>
  17.             <artifactId>spring-boot-starter-web</artifactId>
  18.         </dependency>
  19.         <dependency>
  20.             <groupId>org.springframework.cloud</groupId>
  21.             <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
  22.         </dependency>
  23.         <!--sentinel-->
  24.         <dependency>
  25.             <groupId>com.alibaba.cloud</groupId>
  26.             <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
  27.         </dependency>
  29.         <!--健康监控-->
  30.         <dependency>
  31.             <groupId>org.springframework.boot</groupId>
  32.             <artifactId>spring-boot-starter-actuator</artifactId>
  33.         </dependency>
  35.         <dependency>
  36.             <groupId>com.alibaba.cloud</groupId>
  37.             <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
  38.         </dependency>
  39.         <dependency>
  40.             <groupId>com.alibaba.cloud</groupId>
  41.             <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
  42.         </dependency>
  43.         <!-- mybatis 和SpringBoot 整合-->
  44.         <dependency>
  45.             <groupId>org.mybatis.spring.boot</groupId>
  46.             <artifactId>mybatis-spring-boot-starter</artifactId>
  47.         </dependency>
  49.         <!-- MySQL 驱动 -->
  50.         <dependency>
  51.             <groupId>mysql</groupId>
  52.             <artifactId>mysql-connector-java</artifactId>
  53.         </dependency>
  54.         <dependency>
  55.             <groupId>com.alibaba</groupId>
  56.             <artifactId>druid</artifactId>
  57.         </dependency>
  58.         <!-- jdbc -->
  59.         <dependency>
  60.             <groupId>org.springframework.boot</groupId>
  61.             <artifactId>spring-boot-starter-jdbc</artifactId>
  62.         </dependency>
  64.         <dependency>
  65.             <groupId>org.springframework.boot</groupId>
  66.             <artifactId>spring-boot-devtools</artifactId>
  67.             <scope>runtime</scope>
  68.             <optional>true</optional>
  69.         </dependency>
  71.         <dependency>
  72.             <groupId>org.projectlombok</groupId>
  73.             <artifactId>lombok</artifactId>
  74.         </dependency>
  76.         <dependency>
  77.             <groupId>org.springframework.boot</groupId>
  78.             <artifactId>spring-boot-starter-test</artifactId>
  79.             <scope>test</scope>
  80.         </dependency>
  81.         <dependency>
  82.             <groupId>junit</groupId>
  83.             <artifactId>junit</artifactId>
  84.         </dependency>
  85.     </dependencies>
  87. </project>

yaml配置

  1. server:
  2.   port: 8003
  4. spring:
  5.   application:
  6.     name: demo-stream-rabbit-provider
  7.   datasource:
  8.     type: com.alibaba.druid.pool.DruidDataSource
  9.     driver-class-name: com.mysql.jdbc.Driver
  10.     url: jdbc:mysql://192.168.2.20:30569/test?useUnicode=true&characterEncoding=utf-8&useSSL=false
  11.     username: root
  12.     password: root
  13.   cloud:
  14.     nacos:
  15.       server-addr: 192.168.2.20:30010
  16.       username: nacos
  17.       password: nacos
  18.     sentinel:
  19.       transport:
  20.         dashboard: 192.168.2.20:31596    # 控制台的安装位置
  21.     #        port: 8719    # 与sentinel单独连接的端口
  22.     #        client-ip: 192.168.2.6    # 本机的ip，如果sentinel装在虚拟机，必须配这个
  23.     #        port: 30195    # 与sentinel单独连接的端口
  24.     #        client-ip: 192.168.2.20    # 本机的ip，如果sentinel装在虚拟机，必须配这个
  25.     stream:
  26.       binders: # 在此处配置要绑定的rabbitmq的服务信息；
  27.         defaultRabbit: # 表示定义的名称，用于于binding整合
  28.           type: rabbit # 消息组件类型
  29. #          environment: # 设置rabbitmq的相关的环境配置
  30. #            spring:
  31. #              rabbitmq:
  32. #                host: 192.168.2.20
  33. #                port: 31672
  34. #                username: user
  35. #                password: ze2tgb2WGC
  37.       bindings: # 服务的整合处理
  38.         output: # 这个名字是一个通道的名称
  39.           destination: studyExchange # 表示要使用的Exchange名称定义
  40.           content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
  41.           binder: defaultRabbit  # 设置要绑定的消息服务的具体设置
  42.           group: default-group
  43.         input: # 这个名字是一个通道的名称
  44.           destination: studyExchange # 表示要使用的Exchange名称定义
  45.           binder: defaultRabbit  # 设置要绑定的消息服务的具体设置
  46.           group: default-group
  48.   rabbitmq:
  49.     host: 192.168.2.20
  50.     port: 31672
  51.     username: user
  52.     password: ze2tgb2WGC
  55. # 暴露应用信息
  56. management:
  57.   endpoints:
  58.     web:
  59.       exposure:
  60.         include: '*'

示例：controller

  1. package com.test.rabbit.consumer.controller;
  3. import com.test.rabbit.consumer.service.MessageProviderService;
  4. import lombok.RequiredArgsConstructor;
  5. import org.springframework.web.bind.annotation.GetMapping;
  6. import org.springframework.web.bind.annotation.RequestMapping;
  7. import org.springframework.web.bind.annotation.RestController;
  9. /**
  10.  * @author jia
  11.  * @since 2022-04-05 11:39
  12.  */
  13. @RestController
  14. @RequestMapping("/rabbit")
  15. @RequiredArgsConstructor
  16. public class SendMessageController {
  17.     protected final MessageProviderService messageProviderService;
  19.     @GetMapping("/sendMessage")
  20.     public String sendMessage() {
  21.         return messageProviderService.send();
  22.     }
  23. }

示例：service

  1. package com.test.rabbit.consumer.serviceimpl;
  3. import com.test.rabbit.consumer.service.MessageProviderService;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.cloud.stream.annotation.EnableBinding;
  6. import org.springframework.cloud.stream.messaging.Source;
  7. import org.springframework.messaging.MessageChannel;
  8. import org.springframework.messaging.support.MessageBuilder;
  10. import javax.annotation.Resource;
  11. import java.util.UUID;
  13. /**
  14.  * @author jia
  15.  * @since 2022-04-05 11:42
  16.  */
  17. @EnableBinding(Source.class)
  18. @Slf4j
  19. public class MessageProviderServiceImpl implements MessageProviderService {
  20.     @Resource
  21.     MessageChannel output;
  23.     @Override
  24.     public String send() {
  25.         String serial = UUID.randomUUID().toString();
  26.         output.send(MessageBuilder.withPayload(serial).build());
  27.         log.info("***** serial: {}", serial);
  28.         return null;
  29.     }
  30. }

其它

使用原生方式

  1. Properties props = new Properties();
  2. props.put("bootstrap.servers", "localhost:9092");
  3. props.put("transactional.id", "my-transactional-id");
  4. Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
  6. producer.initTransactions();
  8. try {
  9.     producer.beginTransaction();
  10.     for (int i = 0; i < 100; i++)
  11.         producer.send(new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i)));
  12.     producer.commitTransaction();
  13. } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
  14.     // We can't recover from these exceptions, so our only option is to close the producer and exit.
  15.     producer.close();
  16. } catch (KafkaException e) {
  17.     // For all other exceptions, just abort the transaction and try again.
  18.     producer.abortTransaction();
  19. }
  20. producer.close();

参考

https://www.orchome.com/303