Merge remote-tracking branch 'origin/master'
# Conflicts: # dev-protocol-springcloud/SpringCloud项目介绍.md # dev-protocol-springcloud/dev-protocol-springcloud-hystrix/NetflixHystrix.md # dev-protocol-springcloud/dev-protocol-springcloud-hystrix/src/main/java/org/example/controller/HystrixController.javamaster
q
1 week ago
commit
234568cfdc
@ -0,0 +1,3 @@
|
||||
## hystrix-dashboard
|
||||
|
||||
- hystrix 监控面板
|
||||
@ -0,0 +1,21 @@
|
||||
package org.example;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
|
||||
import org.springframework.cloud.netflix.hystrix.dashboard.EnableHystrixDashboard;
|
||||
|
||||
/**
|
||||
* <h1>hystrix dashboard 入口</h1>
|
||||
* 控制台页面 http://127.0.0.1:9999/dev-protocol-springcloud-hystrix-dashboard/hystrix/
|
||||
* 需要监控的服务 stream 拼接: http://127.0.0.1:8111/dev-protocol-springcloud-hystrix/actuator/hystrix.stream
|
||||
* */
|
||||
@EnableDiscoveryClient
|
||||
@SpringBootApplication
|
||||
@EnableHystrixDashboard // 开启 Hystrix Dashboard
|
||||
public class HystrixDashboardApplication {
|
||||
public static void main(String[] args) {
|
||||
|
||||
SpringApplication.run(HystrixDashboardApplication.class, args);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,33 @@
|
||||
server:
|
||||
port: 9999
|
||||
servlet:
|
||||
context-path: /dev-protocol-springcloud-hystrix-dashboard
|
||||
|
||||
spring:
|
||||
application:
|
||||
name: dev-protocol-springcloud-hystrix-dashboard
|
||||
cloud:
|
||||
nacos:
|
||||
# 服务注册发现
|
||||
discovery:
|
||||
enabled: true # 如果不想使用 Nacos 进行服务注册和发现, 设置为 false 即可
|
||||
server-addr: 127.0.0.1:8848
|
||||
# server-addr: 127.0.0.1:8848,127.0.0.1:8849,127.0.0.1:8850 # Nacos 服务器地址
|
||||
namespace: 1ccc74ae-9398-4dbe-b9d7-4f9addf9f40c
|
||||
metadata:
|
||||
management:
|
||||
context-path: ${server.servlet.context-path}/actuator
|
||||
|
||||
hystrix:
|
||||
dashboard:
|
||||
proxy-stream-allow-list: "127.0.0.1"
|
||||
|
||||
# 暴露端点
|
||||
management:
|
||||
endpoints:
|
||||
web:
|
||||
exposure:
|
||||
include: '*'
|
||||
endpoint:
|
||||
health:
|
||||
show-details: always
|
||||
@ -0,0 +1,31 @@
|
||||
### 获取 Token
|
||||
POST http://127.0.0.1:9001/dev-protocol-springcloud-gateway/dev-protocol-springcloud-hystrix/communication/token-by-feign
|
||||
Content-Type: application/json
|
||||
e-commerce-user: eyJhbGciOiJSUzI1NiJ9.eyJlLWNvbW1lcmNlLXVzZXIiOiJ7XCJpZFwiOjEwLFwidXNlcm5hbWVcIjpcIlFpbnlpQGltb29jLmNvbVwifSIsImp0aSI6IjJlMTgwYzA2LWIwZGQtNDJjYy1iNzk5LTZiNGI4N2Q3NTM3NCIsImV4cCI6MTYyNTQxNDQwMH0.P8gKWcwiibEmq80idCl9QzUaqJnQ7FVQ5yxzlMLfRNcG3nWMYh0o1kBO2vKvDGG8BofQ4__xn210R2P7zjH-uEn7hic0jbBt97vLkzvATnFegxaDSjUp9At9BIGl9Ecu7RPCwSQJFdw6NBMBz02Uom8yvHpUJEfv8_jZRurBp94tzES6IAACKC6NFq-3_mQfT4o_zA51U4hBbIBw6qNXvMSqK-VtaqkTNFfJsP_8Y0nmQck0jxmgdtLfRVfZCCp3rhcLNDBX37THQI6U7lvNFETceAKyJ5NFaA8p68lw_3D2IiZER1jEU3M2QIMQGeU0pmQ6IPwGDdA0MP5IKKzj8A
|
||||
token: imooc
|
||||
|
||||
{
|
||||
"username": "Qinyi@imooc.com",
|
||||
"password": "25d55ad283aa400af464c76d713c07ad"
|
||||
}
|
||||
|
||||
|
||||
### 根据提供的 serviceId 获取实例信息
|
||||
GET http://127.0.0.1:9001/dev-protocol-springcloud-gateway/ecommerce-nacos-client/hystrix/hystrix-command-annotation?serviceId=e-commerce-nacos-client
|
||||
Content-Type: application/json
|
||||
e-commerce-user: eyJhbGciOiJSUzI1NiJ9.eyJlLWNvbW1lcmNlLXVzZXIiOiJ7XCJpZFwiOjEwLFwidXNlcm5hbWVcIjpcIlFpbnlpQGltb29jLmNvbVwifSIsImp0aSI6IjJlMTgwYzA2LWIwZGQtNDJjYy1iNzk5LTZiNGI4N2Q3NTM3NCIsImV4cCI6MTYyNTQxNDQwMH0.P8gKWcwiibEmq80idCl9QzUaqJnQ7FVQ5yxzlMLfRNcG3nWMYh0o1kBO2vKvDGG8BofQ4__xn210R2P7zjH-uEn7hic0jbBt97vLkzvATnFegxaDSjUp9At9BIGl9Ecu7RPCwSQJFdw6NBMBz02Uom8yvHpUJEfv8_jZRurBp94tzES6IAACKC6NFq-3_mQfT4o_zA51U4hBbIBw6qNXvMSqK-VtaqkTNFfJsP_8Y0nmQck0jxmgdtLfRVfZCCp3rhcLNDBX37THQI6U7lvNFETceAKyJ5NFaA8p68lw_3D2IiZER1jEU3M2QIMQGeU0pmQ6IPwGDdA0MP5IKKzj8A
|
||||
token: imooc
|
||||
|
||||
|
||||
### 根据提供的 serviceId 获取实例信息
|
||||
GET http://127.0.0.1:9001/dev-protocol-springcloud-gateway/ecommerce-nacos-client/hystrix/simple-hystrix-command?serviceId=e-commerce-nacos-client
|
||||
Content-Type: application/json
|
||||
e-commerce-user: eyJhbGciOiJSUzI1NiJ9.eyJlLWNvbW1lcmNlLXVzZXIiOiJ7XCJpZFwiOjEwLFwidXNlcm5hbWVcIjpcIlFpbnlpQGltb29jLmNvbVwifSIsImp0aSI6IjJlMTgwYzA2LWIwZGQtNDJjYy1iNzk5LTZiNGI4N2Q3NTM3NCIsImV4cCI6MTYyNTQxNDQwMH0.P8gKWcwiibEmq80idCl9QzUaqJnQ7FVQ5yxzlMLfRNcG3nWMYh0o1kBO2vKvDGG8BofQ4__xn210R2P7zjH-uEn7hic0jbBt97vLkzvATnFegxaDSjUp9At9BIGl9Ecu7RPCwSQJFdw6NBMBz02Uom8yvHpUJEfv8_jZRurBp94tzES6IAACKC6NFq-3_mQfT4o_zA51U4hBbIBw6qNXvMSqK-VtaqkTNFfJsP_8Y0nmQck0jxmgdtLfRVfZCCp3rhcLNDBX37THQI6U7lvNFETceAKyJ5NFaA8p68lw_3D2IiZER1jEU3M2QIMQGeU0pmQ6IPwGDdA0MP5IKKzj8A
|
||||
token: imooc
|
||||
|
||||
|
||||
### 根据提供的 serviceId 获取实例信息
|
||||
GET http://127.0.0.1:9001/dev-protocol-springcloud-gateway/ecommerce-nacos-client/hystrix/hystrix-observable-command?serviceId=e-commerce-nacos-client
|
||||
Content-Type: application/json
|
||||
e-commerce-user: eyJhbGciOiJSUzI1NiJ9.eyJlLWNvbW1lcmNlLXVzZXIiOiJ7XCJpZFwiOjEwLFwidXNlcm5hbWVcIjpcIlFpbnlpQGltb29jLmNvbVwifSIsImp0aSI6IjJlMTgwYzA2LWIwZGQtNDJjYy1iNzk5LTZiNGI4N2Q3NTM3NCIsImV4cCI6MTYyNTQxNDQwMH0.P8gKWcwiibEmq80idCl9QzUaqJnQ7FVQ5yxzlMLfRNcG3nWMYh0o1kBO2vKvDGG8BofQ4__xn210R2P7zjH-uEn7hic0jbBt97vLkzvATnFegxaDSjUp9At9BIGl9Ecu7RPCwSQJFdw6NBMBz02Uom8yvHpUJEfv8_jZRurBp94tzES6IAACKC6NFq-3_mQfT4o_zA51U4hBbIBw6qNXvMSqK-VtaqkTNFfJsP_8Y0nmQck0jxmgdtLfRVfZCCp3rhcLNDBX37THQI6U7lvNFETceAKyJ5NFaA8p68lw_3D2IiZER1jEU3M2QIMQGeU0pmQ6IPwGDdA0MP5IKKzj8A
|
||||
token: imooc
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 133 KiB |
@ -0,0 +1,50 @@
|
||||
package org.example.hystrix.request_merge;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.netflix.hystrix.HystrixCommand;
|
||||
import com.netflix.hystrix.HystrixCommandGroupKey;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.example.service.NacosClientService4HystrixDemo;
|
||||
import org.springframework.cloud.client.ServiceInstance;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* <h1>批量请求 Hystrix Command</h1>
|
||||
* */
|
||||
@Slf4j
|
||||
public class NacosClientBatchCommand extends HystrixCommand<List<List<ServiceInstance>>> {
|
||||
|
||||
private final NacosClientService4HystrixDemo nacosClientService;
|
||||
private final List<String> serviceIds;
|
||||
|
||||
protected NacosClientBatchCommand(
|
||||
NacosClientService4HystrixDemo nacosClientService, List<String> serviceIds
|
||||
) {
|
||||
|
||||
super(
|
||||
HystrixCommand.Setter.withGroupKey(
|
||||
HystrixCommandGroupKey.Factory.asKey("NacosClientBatchCommand")
|
||||
)
|
||||
);
|
||||
|
||||
this.nacosClientService = nacosClientService;
|
||||
this.serviceIds = serviceIds;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected List<List<ServiceInstance>> run() throws Exception {
|
||||
|
||||
log.info("use nacos client batch command to get result: [{}]",
|
||||
JSON.toJSONString(serviceIds));
|
||||
return nacosClientService.getNacosClientInfos(serviceIds);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected List<List<ServiceInstance>> getFallback() {
|
||||
|
||||
log.warn("nacos client batch command failure, use fallback");
|
||||
return Collections.emptyList();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,85 @@
|
||||
package org.example.hystrix.request_merge;
|
||||
|
||||
import com.netflix.hystrix.HystrixCollapser;
|
||||
import com.netflix.hystrix.HystrixCollapserKey;
|
||||
import com.netflix.hystrix.HystrixCollapserProperties;
|
||||
import com.netflix.hystrix.HystrixCommand;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.example.service.NacosClientService4HystrixDemo;
|
||||
import org.springframework.cloud.client.ServiceInstance;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* <h1>请求合并器</h1>
|
||||
* */
|
||||
@Slf4j
|
||||
public class NacosClientCollapseCommand
|
||||
extends HystrixCollapser<List<List<ServiceInstance>>, List<ServiceInstance>, String> {
|
||||
// 批量返回类型, 单个请求对象的返回类型, 请求参数的类型
|
||||
|
||||
private final NacosClientService4HystrixDemo nacosClientService;
|
||||
private final String serviceId;
|
||||
|
||||
public NacosClientCollapseCommand(NacosClientService4HystrixDemo nacosClientService, String serviceId) {
|
||||
|
||||
super(
|
||||
HystrixCollapser.Setter.withCollapserKey(
|
||||
HystrixCollapserKey.Factory.asKey("NacosClientCollapseCommand")
|
||||
).andCollapserPropertiesDefaults(
|
||||
HystrixCollapserProperties.Setter().withTimerDelayInMilliseconds(300) // 等待 300ms 合并请求
|
||||
)
|
||||
);
|
||||
|
||||
this.nacosClientService = nacosClientService;
|
||||
this.serviceId = serviceId;
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>获取请求中的参数</h2>
|
||||
* */
|
||||
@Override
|
||||
public String getRequestArgument() {
|
||||
return this.serviceId;
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>创建批量请求 Hystrix Command</h2>
|
||||
* */
|
||||
@Override
|
||||
protected HystrixCommand<List<List<ServiceInstance>>> createCommand(
|
||||
Collection<CollapsedRequest<List<ServiceInstance>, String>> collapsedRequests) {
|
||||
|
||||
List<String> serviceIds = new ArrayList<>(collapsedRequests.size()); // 合并请求的大小就是参数们的大小
|
||||
serviceIds.addAll(
|
||||
collapsedRequests.stream()
|
||||
.map(CollapsedRequest::getArgument) // 获取每个请求的参数
|
||||
.collect(Collectors.toList())
|
||||
);
|
||||
|
||||
return new NacosClientBatchCommand(nacosClientService, serviceIds);
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>响应分发给单独的请求</h2>
|
||||
* */
|
||||
@Override
|
||||
protected void mapResponseToRequests(List<List<ServiceInstance>> batchResponse,
|
||||
Collection<CollapsedRequest<List<ServiceInstance>,
|
||||
String>> collapsedRequests) {
|
||||
|
||||
int count = 0;
|
||||
|
||||
for (CollapsedRequest<List<ServiceInstance>, String> collapsedRequest : collapsedRequests) {
|
||||
|
||||
// 从批量响应集合中按顺序取出结果
|
||||
List<ServiceInstance> instances = batchResponse.get(count++);
|
||||
// 将结果返回原 Response 中
|
||||
collapsedRequest.setResponse(instances);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,31 @@
|
||||
package org.example.service;
|
||||
|
||||
import org.example.service.hystrix.bak.AuthorityFeignClientFallback;
|
||||
import org.example.service.hystrix.bak.AuthorityFeignClientFallbackFactory;
|
||||
import org.springframework.cloud.openfeign.FeignClient;
|
||||
import org.springframework.web.bind.annotation.RequestBody;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RequestMethod;
|
||||
|
||||
/**
|
||||
* <h1>与 Authority 服务通信的 Feign Client 接口定义</h1>
|
||||
* */
|
||||
@FeignClient(
|
||||
// contextId 是对 FeignClient 的声明, 每一个进行的通信都要进行定义, value 表示需要进行通信的服务id是什么
|
||||
contextId = "AuthorityFeignClient", value = "e-commerce-authority-center",
|
||||
// fallback = AuthorityFeignClientFallback.class, // fixme fallback 和 fallbackFactory 只能使用一个
|
||||
fallbackFactory = AuthorityFeignClientFallbackFactory.class
|
||||
)
|
||||
public interface AuthorityFeignClientHystrixDemo {
|
||||
|
||||
/**
|
||||
* <h2>通过 OpenFeign 访问 Authority 获取 Token</h2>
|
||||
*
|
||||
* value 只需要定义路径即可, 不需要定义 ip + port
|
||||
* consumes, produces: 标识请求和返回的数据格式, 可以不指定, 但是在使用原生 Api 接口的时候必须要进行指定, 要不会不能进行识别对应的接口
|
||||
* */
|
||||
@RequestMapping(value = "/ecommerce-authority-center/authority/token", // fixme ：这里的调用代码是不对的， 仅仅作为 Hystrix 举例使用
|
||||
method = RequestMethod.POST,
|
||||
consumes = "application/json", produces = "application/json")
|
||||
String getTokenByFeign(@RequestBody String usernameAndPassword);
|
||||
}
|
||||
@ -0,0 +1,21 @@
|
||||
package org.example.service.hystrix.bak;
|
||||
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.example.service.AuthorityFeignClientHystrixDemo;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* <h1>AuthorityFeignClient 后备 fallback</h1>
|
||||
* */
|
||||
@Slf4j
|
||||
@Component
|
||||
public class AuthorityFeignClientFallback implements AuthorityFeignClientHystrixDemo {
|
||||
@Override
|
||||
public String getTokenByFeign(String usernameAndPassword) {
|
||||
log.info("authority feign client get token by feign request error " +
|
||||
"(Hystrix Fallback): [{}]", JSON.toJSONString(usernameAndPassword));
|
||||
return "qqqqq";
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,27 @@
|
||||
package org.example.service.hystrix.bak;
|
||||
|
||||
import feign.hystrix.FallbackFactory;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.example.service.AuthorityFeignClientHystrixDemo;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* <h1>OpenFeign 集成 Hystrix 的另一种模式</h1>
|
||||
* */
|
||||
@Slf4j
|
||||
@Component
|
||||
public class AuthorityFeignClientFallbackFactory implements FallbackFactory<AuthorityFeignClientHystrixDemo> {
|
||||
@Override
|
||||
public AuthorityFeignClientHystrixDemo create(Throwable throwable) {
|
||||
|
||||
log.warn("authority feign client get token by feign request error " +
|
||||
"(Hystrix FallbackFactory): [{}]", throwable.getMessage(), throwable);
|
||||
return new AuthorityFeignClientHystrixDemo() {
|
||||
|
||||
@Override
|
||||
public String getTokenByFeign(String usernameAndPassword) {
|
||||
return "q-factory";
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,40 @@
|
||||
## 基于 SpringCloud Stream 构建消息驱动微服务
|
||||
|
||||
|
||||
|
||||
### SpringBoot 集成 Kafka 构建消息驱动微服务
|
||||
- 下载安装 kafka
|
||||
- 下载 Kafka :https://kafka.apache.org/quickstart
|
||||
- 解压、启动 ZK 和 Kafka Server 即可(使用默认配置)
|
||||
---
|
||||
- 基本架构
|
||||
- Producer -> Message[Topic] -> Kafka Broker -> Partition <- Consumer[Topic]
|
||||
|
||||
### SpringBoot 集成 RocketMQ 构建消息驱动微服务
|
||||
- 下载、安装 RocketMQ
|
||||
- 下载 RocketMQ: http://rocketmg.apache.org/docs/quick-start
|
||||
- 下载以 bin-release 结尾的 zip 包解压即完成安装
|
||||
---
|
||||
- MQ 的启动, 关注2个
|
||||
- mqnamesrv
|
||||
- sh mqnamesrv
|
||||
- mqbroker
|
||||
- sh mqbroker -n localhost:9876
|
||||
|
||||
### SpringCloud Stream 消息驱动组件概览
|
||||
- 为什么会出现 SpringCloud Stream
|
||||
- 如果没有 SpringCloud Stream,我们会怎么应用消息驱动?
|
||||
- Producer -> Message -> Kafka/RocketMQ <- Consumer
|
||||
- 有了Stream
|
||||
- Producer -> Message -> [Kafka/RocketMQ][Stream] <- Consumer
|
||||
|
||||
### 基于 SpringCloud Stream 消息驱动的简单应用
|
||||
- [dev-protocol-springcloud-stream](..%2Fdev-protocol-springcloud-stream)
|
||||
|
||||
### 自定义 Stream 消息通信信道实现定制分发
|
||||
- [dev-protocol-springcloud-stream](..%2Fdev-protocol-springcloud-stream)
|
||||
|
||||
### SpringCloud Stream 消息分组和消费分区的配置与说明
|
||||
- [dev-protocol-springcloud-stream](..%2Fdev-protocol-springcloud-stream)
|
||||
|
||||
### SpringCloud Stream 消息驱动组件总结
|
||||
@ -0,0 +1,99 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>org.example</groupId>
|
||||
<artifactId>dev-protocol</artifactId>
|
||||
<version>1.0-SNAPSHOT</version>
|
||||
<relativePath>../../pom.xml</relativePath>
|
||||
</parent>
|
||||
|
||||
<artifactId>dev-protocol-springcloud-message-study</artifactId>
|
||||
<version>1.0-SNAPSHOT</version>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<name>dev-protocol-springcloud-message-study</name>
|
||||
<description>MQ 学习实践</description>
|
||||
|
||||
<properties>
|
||||
<maven.compiler.source>8</maven.compiler.source>
|
||||
<maven.compiler.target>8</maven.compiler.target>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-web</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.projectlombok</groupId>
|
||||
<artifactId>lombok</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<!-- SpringBoot 监控端点 -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<!-- 让 SpringBoot 能够识别 bootstrap.yml -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-context</artifactId>
|
||||
<version>2.2.6.RELEASE</version>
|
||||
</dependency>
|
||||
<!-- Kafka -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.kafka</groupId>
|
||||
<artifactId>spring-kafka</artifactId>
|
||||
<version>2.5.0.RELEASE</version>
|
||||
</dependency>
|
||||
<!-- RocketMQ 这个版本必须和你的 RocketMQ 版本匹配 -->
|
||||
<dependency>
|
||||
<groupId>org.apache.rocketmq</groupId>
|
||||
<artifactId>rocketmq-spring-boot-starter</artifactId>
|
||||
<version>2.1.0</version> <!-- 匹配 4.8.0-->
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.commons</groupId>
|
||||
<artifactId>commons-lang3</artifactId>
|
||||
<version>3.11</version>
|
||||
</dependency>
|
||||
<!-- MySQL 驱动 -->
|
||||
<dependency>
|
||||
<groupId>mysql</groupId>
|
||||
<artifactId>mysql-connector-java</artifactId>
|
||||
<version>8.0.19</version>
|
||||
</dependency>
|
||||
<!-- Spring Data Jpa -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-data-jpa</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<finalName>${artifactId}</finalName>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<goals>
|
||||
<goal>repackage</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
@ -0,0 +1,14 @@
|
||||
package org.example;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
/**
|
||||
* <h1>工程启动入口</h1>
|
||||
* */
|
||||
@SpringBootApplication
|
||||
public class Application {
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(Application.class, args);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,41 @@
|
||||
package org.example.controller;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.example.kafka.KafkaProducer;
|
||||
import org.example.vo.QMessage;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RequestParam;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
/**
|
||||
* <h1>SpringBoot 集成 kafka 发送消息</h1>
|
||||
* */
|
||||
@Slf4j
|
||||
@RestController
|
||||
@RequestMapping("/kafka")
|
||||
public class KafkaController {
|
||||
|
||||
private final ObjectMapper mapper;
|
||||
private final KafkaProducer kafkaProducer;
|
||||
|
||||
public KafkaController(ObjectMapper mapper, KafkaProducer kafkaProducer) {
|
||||
this.mapper = mapper;
|
||||
this.kafkaProducer = kafkaProducer;
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>发送 kafka 消息</h2>
|
||||
* */
|
||||
@GetMapping("/send-message")
|
||||
public void sendMessage(@RequestParam(required = false) String key,
|
||||
@RequestParam String topic) throws Exception {
|
||||
|
||||
QMessage message = new QMessage(
|
||||
1,
|
||||
"q-Study-Message"
|
||||
);
|
||||
kafkaProducer.sendMessage(key, mapper.writeValueAsString(message), topic);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,51 @@
|
||||
package org.example.controller;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.example.rocket.RocketMQProducer;
|
||||
import org.example.vo.QMessage;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
/**
|
||||
* <h1>SpringBoot 集成 RocketMQ</h1>
|
||||
* */
|
||||
@Slf4j
|
||||
@RestController
|
||||
@RequestMapping("/rocket-mq")
|
||||
public class RocketMQController {
|
||||
|
||||
private static final QMessage RocketMQMessage = new QMessage(
|
||||
1,
|
||||
"Q-Study-RocketMQ-In-SpringBoot"
|
||||
);
|
||||
|
||||
private final RocketMQProducer rocketMQProducer;
|
||||
|
||||
public RocketMQController(RocketMQProducer rocketMQProducer) {
|
||||
this.rocketMQProducer = rocketMQProducer;
|
||||
}
|
||||
|
||||
@GetMapping("/message-with-value")
|
||||
public void sendMessageWithValue() {
|
||||
rocketMQProducer.sendMessageWithValue(JSON.toJSONString(RocketMQMessage));
|
||||
}
|
||||
|
||||
@GetMapping("/message-with-key")
|
||||
public void sendMessageWithKey() {
|
||||
rocketMQProducer.sendMessageWithKey("qy", JSON.toJSONString(RocketMQMessage));
|
||||
}
|
||||
|
||||
@GetMapping("/message-with-tag")
|
||||
public void sendMessageWithTag() {
|
||||
rocketMQProducer.sendMessageWithTag("qy",
|
||||
JSON.toJSONString(RocketMQMessage));
|
||||
}
|
||||
|
||||
@GetMapping("/message-with-all")
|
||||
public void sendMessageWithAll() {
|
||||
rocketMQProducer.sendMessageWithAll("q", "q",
|
||||
JSON.toJSONString(RocketMQMessage));
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,81 @@
|
||||
package org.example.kafka;
|
||||
|
||||
import org.apache.kafka.clients.consumer.ConsumerConfig;
|
||||
import org.apache.kafka.clients.producer.ProducerConfig;
|
||||
import org.apache.kafka.common.serialization.StringDeserializer;
|
||||
import org.apache.kafka.common.serialization.StringSerializer;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
|
||||
import org.springframework.kafka.core.*;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* <h1>通过代码自定义 Kafka 配置</h1>
|
||||
*
|
||||
* 复杂不会怎么更改的配置信息最好以代码的方式进行配置
|
||||
* 一般一些常用的改动的配置, 直接放在配置文件中
|
||||
* */
|
||||
@Configuration
|
||||
public class KafkaConfig {
|
||||
|
||||
@Value("${spring.kafka.bootstrap-servers}")
|
||||
private String bootstrapServers;
|
||||
|
||||
/**
|
||||
* <h2>Kafka Producer 工厂类配置</h2>
|
||||
* */
|
||||
@Bean
|
||||
public ProducerFactory<String, String> producerFactory() {
|
||||
|
||||
Map<String, Object> configs = new HashMap<>();
|
||||
configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
|
||||
configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
|
||||
configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
|
||||
|
||||
return new DefaultKafkaProducerFactory<>(configs);
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>Kafka Producer 客户端</h2>
|
||||
* */
|
||||
@Bean
|
||||
public KafkaTemplate<String, String> kafkaTemplate() {
|
||||
return new KafkaTemplate<>(producerFactory());
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>Kafka Consumer 工厂类配置</h2>
|
||||
* */
|
||||
@Bean
|
||||
public ConsumerFactory<String, String> consumerFactory() {
|
||||
|
||||
Map<String, Object> props = new HashMap<>();
|
||||
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
|
||||
// 因为配置为拉取模式, 最多拉取 50条记录
|
||||
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
|
||||
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
|
||||
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
|
||||
|
||||
return new DefaultKafkaConsumerFactory<>(props);
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>Kafka Consumer 监听器工厂类配置</h2>
|
||||
* */
|
||||
@Bean
|
||||
public ConcurrentKafkaListenerContainerFactory<String, String>
|
||||
kafkaListenerContainerFactory() {
|
||||
|
||||
ConcurrentKafkaListenerContainerFactory<String, String> factory =
|
||||
new ConcurrentKafkaListenerContainerFactory<>();
|
||||
// 并发数就是一个消费者实例起几个线程
|
||||
factory.setConcurrency(3);
|
||||
factory.setConsumerFactory(consumerFactory());
|
||||
|
||||
return factory;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,56 @@
|
||||
package org.example.kafka;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.example.vo.QMessage;
|
||||
import org.springframework.kafka.annotation.KafkaListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.Optional;
|
||||
|
||||
/**
|
||||
* <h1>Kafka 消费者</h1>
|
||||
* */
|
||||
@Slf4j
|
||||
@Component
|
||||
public class KafkaConsumer {
|
||||
|
||||
private final ObjectMapper mapper;
|
||||
|
||||
public KafkaConsumer(ObjectMapper mapper) {
|
||||
this.mapper = mapper;
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>监听 Kafka 消息并消费</h2>
|
||||
* */
|
||||
@KafkaListener(topics = {"qinyi-springboot"}, groupId = "qinyi-springboot-kafka")
|
||||
public void listener01(ConsumerRecord<String, String> record) throws Exception {
|
||||
|
||||
String key = record.key();
|
||||
String value = record.value();
|
||||
|
||||
QMessage kafkaMessage = mapper.readValue(value, QMessage.class);
|
||||
log.info("in listener01 consume kafka message: [{}], [{}]",
|
||||
key, mapper.writeValueAsString(kafkaMessage));
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>监听 Kafka 消息并消费</h2>
|
||||
* 不知道发送的类是什么类型的时候发送的
|
||||
* */
|
||||
@KafkaListener(topics = {"qinyi-springboot"}, groupId = "qinyi-springboot-kafka-1")
|
||||
public void listener02(ConsumerRecord<?, ?> record) throws Exception {
|
||||
|
||||
Optional<?> _kafkaMessage = Optional.ofNullable(record.value());
|
||||
if (_kafkaMessage.isPresent()) {
|
||||
Object message = _kafkaMessage.get();
|
||||
// 如果不能确定类型的时候, 下面的代码要进行包装
|
||||
QMessage kafkaMessage = mapper.readValue(message.toString(),
|
||||
QMessage.class);
|
||||
log.info("in listener02 consume kafka message: [{}]",
|
||||
mapper.writeValueAsString(kafkaMessage));
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,77 @@
|
||||
package org.example.kafka;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.springframework.kafka.core.KafkaTemplate;
|
||||
import org.springframework.kafka.support.SendResult;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.util.concurrent.ListenableFuture;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* <h1>kafka 生产者</h1>
|
||||
* */
|
||||
@Slf4j
|
||||
@Component
|
||||
public class KafkaProducer {
|
||||
|
||||
private final KafkaTemplate<String, String> kafkaTemplate;
|
||||
|
||||
public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
|
||||
this.kafkaTemplate = kafkaTemplate;
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>发送 kafka 消息</h2>
|
||||
* */
|
||||
public void sendMessage(String key, String value, String topic) {
|
||||
|
||||
if (StringUtils.isBlank(value) || StringUtils.isBlank(topic)) {
|
||||
throw new IllegalArgumentException("value or topic is null or empty");
|
||||
}
|
||||
|
||||
ListenableFuture<SendResult<String, String>> future = StringUtils.isBlank(key) ?
|
||||
kafkaTemplate.send(topic, value) : kafkaTemplate.send(topic, key, value);
|
||||
|
||||
// 异步回调的方式获取通知
|
||||
future.addCallback(
|
||||
success -> {
|
||||
// 元数据信息不为空
|
||||
assert null != success && null != success.getRecordMetadata();
|
||||
// 发送到 kafka 的 topic
|
||||
String _topic = success.getRecordMetadata().topic();
|
||||
// 消息发送到的分区
|
||||
int partition = success.getRecordMetadata().partition();
|
||||
// 消息在分区内的 offset
|
||||
long offset = success.getRecordMetadata().offset();
|
||||
|
||||
log.info("send kafka message success: [{}], [{}], [{}]",
|
||||
_topic, partition, offset);
|
||||
}, failure -> {
|
||||
log.error("send kafka message failure: [{}], [{}], [{}]",
|
||||
key, value, topic);
|
||||
}
|
||||
);
|
||||
// future 支持多次获取消息, 不需要重新发送消息
|
||||
|
||||
// 同步等待的方式获取通知
|
||||
try {
|
||||
// SendResult<String, String> sendResult = future.get();
|
||||
SendResult<String, String> sendResult = future.get(5, TimeUnit.SECONDS);
|
||||
|
||||
// 发送到 kafka 的 topic
|
||||
String _topic = sendResult.getRecordMetadata().topic();
|
||||
// 消息发送到的分区
|
||||
int partition = sendResult.getRecordMetadata().partition();
|
||||
// 消息在分区内的 offset
|
||||
long offset = sendResult.getRecordMetadata().offset();
|
||||
|
||||
log.info("send kafka message success: [{}], [{}], [{}]",
|
||||
_topic, partition, offset);
|
||||
} catch (Exception ex) {
|
||||
log.error("send kafka message failure: [{}], [{}], [{}]",
|
||||
key, value, topic);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,29 @@
|
||||
package org.example.rocket;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.rocketmq.common.message.MessageExt;
|
||||
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
|
||||
import org.apache.rocketmq.spring.core.RocketMQListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* <h1>第三个 RocketMQ 消费者, </h1>
|
||||
* */
|
||||
@Slf4j
|
||||
@Component
|
||||
@RocketMQMessageListener(
|
||||
topic = "q-study-rocketmq",
|
||||
consumerGroup = "q-springboot-rocketmq-message-ext"
|
||||
)
|
||||
public class RocketMQConsumerMessageExt implements RocketMQListener<MessageExt> {
|
||||
|
||||
@Override
|
||||
public void onMessage(MessageExt message) {
|
||||
|
||||
String value = new String(message.getBody());
|
||||
log.info("consume message in RocketMQConsumerMessageExt: [{}], [{}]",
|
||||
message.getKeys(), value); // 能拿到消息中的 key
|
||||
log.info("MessageExt: [{}]", JSON.toJSONString(message)); // 会慢一些
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,27 @@
|
||||
package org.example.rocket;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
|
||||
import org.apache.rocketmq.spring.core.RocketMQListener;
|
||||
import org.example.vo.QMessage;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* <h1>第四个, RocketMQ 消费者, 指定消费带有 tag 的消息, 且消费的是 Java Pojo</h1>
|
||||
* */
|
||||
@Slf4j
|
||||
@Component
|
||||
@RocketMQMessageListener(
|
||||
topic = "imooc-study-rocketmq",
|
||||
consumerGroup = "q-springboot-rocketmq-tag-object",
|
||||
selectorExpression = "q" // 根据 tag 做过滤
|
||||
)
|
||||
public class RocketMQConsumerObject implements RocketMQListener<QMessage> {
|
||||
@Override
|
||||
public void onMessage(QMessage message) {
|
||||
log.info("consume message in RocketMQConsumerObject: [{}]",
|
||||
JSON.toJSONString(message));
|
||||
// so something
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,28 @@
|
||||
package org.example.rocket;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
|
||||
import org.apache.rocketmq.spring.core.RocketMQListener;
|
||||
import org.example.vo.QMessage;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* <h1>第一个 RocketMQ 消费者</h1>
|
||||
* */
|
||||
@Slf4j
|
||||
@Component
|
||||
@RocketMQMessageListener(
|
||||
topic = "q-study-rocketmq",
|
||||
consumerGroup = "q-springboot-rocketmq-string"
|
||||
)
|
||||
public class RocketMQConsumerString implements RocketMQListener<String> {
|
||||
|
||||
@Override
|
||||
public void onMessage(String message) {
|
||||
|
||||
QMessage rocketMessage = JSON.parseObject(message, QMessage.class);
|
||||
log.info("consume message in RocketMQConsumerString: [{}]",
|
||||
JSON.toJSONString(rocketMessage));
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,28 @@
|
||||
package org.example.rocket;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
|
||||
import org.apache.rocketmq.spring.core.RocketMQListener;
|
||||
import org.example.vo.QMessage;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* <h1>第二个 RocketMQ 消费者, 指定了消费带有 tag 的消息</h1>
|
||||
* */
|
||||
@Slf4j
|
||||
@Component
|
||||
@RocketMQMessageListener(
|
||||
topic = "q-study-rocketmq",
|
||||
consumerGroup = "q-springboot-rocketmq-tag-string",
|
||||
selectorExpression = "qy" // 根据 tag 过滤, tag 中要带有 qy
|
||||
)
|
||||
public class RocketMQConsumerTagString implements RocketMQListener<String> {
|
||||
@Override
|
||||
public void onMessage(String message) {
|
||||
|
||||
QMessage rocketMessage = JSON.parseObject(message, QMessage.class);
|
||||
log.info("consume message in RocketMQConsumerTagString: [{}]",
|
||||
JSON.toJSONString(rocketMessage));
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,100 @@
|
||||
package org.example.rocket;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.rocketmq.client.producer.SendCallback;
|
||||
import org.apache.rocketmq.client.producer.SendResult;
|
||||
import org.apache.rocketmq.spring.core.RocketMQTemplate;
|
||||
import org.apache.rocketmq.spring.support.RocketMQHeaders;
|
||||
import org.example.vo.QMessage;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* <h1>通过 RocketMQ 发送消息</h1>
|
||||
* Spring Messaging 模块
|
||||
* */
|
||||
@Slf4j
|
||||
@Component
|
||||
public class RocketMQProducer {
|
||||
|
||||
/** 类似 Kafka 中的 topic, 默认的读写队列都是4个, 默认自动创建topic */
|
||||
private static final String TOPIC = "q-study-rocketmq";
|
||||
|
||||
/** RocketMQ 客户端 */
|
||||
private final RocketMQTemplate rocketMQTemplate;
|
||||
|
||||
public RocketMQProducer(RocketMQTemplate rocketMQTemplate) {
|
||||
this.rocketMQTemplate = rocketMQTemplate;
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>使用同步的方式发送消息, 不指定 key 和 tag</h2>
|
||||
* */
|
||||
public void sendMessageWithValue(String value) {
|
||||
|
||||
// 随机选择一个 Topic 的 Message Queue 发送消息
|
||||
SendResult sendResult = rocketMQTemplate.syncSend(TOPIC, value);
|
||||
log.info("sendMessageWithValue result: [{}]", JSON.toJSONString(sendResult));
|
||||
|
||||
SendResult sendResultOrderly = rocketMQTemplate.syncSendOrderly(
|
||||
TOPIC, value, "QQQ"
|
||||
);
|
||||
log.info("sendMessageWithValue orderly result: [{}]",
|
||||
JSON.toJSONString(sendResultOrderly));
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>使用异步的方式发送消息, 指定 key</h2>
|
||||
* */
|
||||
public void sendMessageWithKey(String key, String value) {
|
||||
|
||||
Message<String> message = MessageBuilder.withPayload(value)
|
||||
// 这个Key 不是分区的效果, 只是方便进行查询, 在设置的时候, 可以使用空格进行分开, 例如: aaaa bbb
|
||||
.setHeader(RocketMQHeaders.KEYS, key).build();
|
||||
|
||||
// 异步发送消息, 并设定回调
|
||||
rocketMQTemplate.asyncSend(TOPIC, message, new SendCallback() {
|
||||
|
||||
@Override
|
||||
public void onSuccess(SendResult sendResult) {
|
||||
log.info("sendMessageWithKey success result: [{}]",
|
||||
JSON.toJSONString(sendResult));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onException(Throwable e) {
|
||||
log.error("sendMessageWithKey failure: [{}]", e.getMessage(), e);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>使用同步的方式发送消息, 带有 tag, 且发送的是 Java Pojo</h2>
|
||||
* 发送消息可以定义自己的消息数据结构
|
||||
* */
|
||||
public void sendMessageWithTag(String tag, String value) {
|
||||
|
||||
QMessage qMessage = JSON.parseObject(value, QMessage.class);
|
||||
SendResult sendResult = rocketMQTemplate.syncSend(
|
||||
String.format("%s:%s", TOPIC, tag), // 不同的消费者组使用不同的 tag
|
||||
qMessage
|
||||
);
|
||||
log.info("sendMessageWithTag result: [{}]", JSON.toJSONString(sendResult));
|
||||
}
|
||||
|
||||
/**
|
||||
* <h2>使用同步的方式发送消息, 带有 key 和 tag</h2>
|
||||
* */
|
||||
public void sendMessageWithAll(String key, String tag, String value) {
|
||||
|
||||
Message<String> message = MessageBuilder.withPayload(value)
|
||||
.setHeader(RocketMQHeaders.KEYS, key).build();
|
||||
SendResult sendResult = rocketMQTemplate.syncSend(
|
||||
String.format("%s:%s", TOPIC, tag),
|
||||
message
|
||||
);
|
||||
log.info("sendMessageWithAll result: [{}]", JSON.toJSONString(sendResult));
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,27 @@
|
||||
package org.example.vo;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
/**
|
||||
* <h1>通过 Kafka 传递的消息对象</h1>
|
||||
* */
|
||||
@Data
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class QMessage {
|
||||
|
||||
/**
|
||||
* 唯一的标识一个消息对象
|
||||
*/
|
||||
private Integer id;
|
||||
|
||||
/**
|
||||
* 项目名称
|
||||
*/
|
||||
private String projectName;
|
||||
|
||||
// todo 自己进行扩展
|
||||
|
||||
}
|
||||
@ -0,0 +1,49 @@
|
||||
server:
|
||||
port: 8001
|
||||
servlet:
|
||||
context-path: /dev-protocol-springcloud-message-study
|
||||
|
||||
spring:
|
||||
# SpringBoot 集成 Kafka 的配置, 最低配置只需要配置 spring.kafka.bootstrap-servers
|
||||
kafka:
|
||||
bootstrap-servers: 127.0.0.1:9092
|
||||
jpa:
|
||||
show-sql: true
|
||||
hibernate:
|
||||
ddl-auto: none
|
||||
properties:
|
||||
hibernate.show_sql: true
|
||||
hibernate.format_sql: true
|
||||
open-in-view: false
|
||||
datasource:
|
||||
# 数据源
|
||||
url: jdbc:mysql://127.0.0.1:3306/dev_protocol_springcloud_project?autoReconnect=true&useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=GMT%2B8
|
||||
username: root
|
||||
password: root
|
||||
type: com.zaxxer.hikari.HikariDataSource
|
||||
driver-class-name: com.mysql.cj.jdbc.Driver
|
||||
# 连接池
|
||||
hikari:
|
||||
maximum-pool-size: 8
|
||||
minimum-idle: 4
|
||||
idle-timeout: 30000
|
||||
connection-timeout: 30000
|
||||
max-lifetime: 45000
|
||||
auto-commit: true
|
||||
pool-name: devProtocolSpringcloudHikariCP
|
||||
# consumer:
|
||||
# 如果 Consumer 没有指定 group-id, 则使用配置文件中配置的; 如果配置文件中也没有定义, 则由框架随机生成
|
||||
# group-id: imooc-study-ecommerce
|
||||
# auto-offset-reset: latest
|
||||
# key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
|
||||
# value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
|
||||
# producer:
|
||||
# key-serializer: org.apache.kafka.common.serialization.StringSerializer
|
||||
# value-serializer: org.apache.kafka.common.serialization.StringSerializer
|
||||
|
||||
# RocketMQ 的配置, 这是最低配置
|
||||
rocketmq:
|
||||
name-server: 127.0.0.1:9876
|
||||
producer:
|
||||
# 发送同一类消息的设置为同一个 group, 保证唯一
|
||||
group: dev-protocol-springcloud-message-study
|
||||
@ -0,0 +1,16 @@
|
||||
spring:
|
||||
profiles:
|
||||
# prod, dev
|
||||
active: dev
|
||||
application:
|
||||
name: dev-protocol-springcloud-message-study
|
||||
|
||||
# 暴露端点
|
||||
management:
|
||||
endpoints:
|
||||
web:
|
||||
exposure:
|
||||
include: '*'
|
||||
endpoint:
|
||||
health:
|
||||
show-details: always
|
||||
@ -0,0 +1,7 @@
|
||||
### kafka-send-message
|
||||
GET http://127.0.0.1:8001/dev-protocol-springcloud-message-study/kafka/send-message?key=qqq&topic=q-springboot
|
||||
Content-Type: application/json
|
||||
|
||||
### kafka-send-message - (测试kafka 支持无key消息, key 是用来分区的)
|
||||
GET http://127.0.0.1:8001/dev-protocol-springcloud-message-study/kafka/send-message?topic=q-springboot
|
||||
Content-Type: application/json
|
||||
@ -0,0 +1,15 @@
|
||||
### message-with-value
|
||||
GET http://127.0.0.1:8001/dev-protocol-springcloud-message-study/rocket-mq/message-with-value
|
||||
Content-Type: application/json
|
||||
|
||||
### message-with-key
|
||||
GET http://127.0.0.1:8001/dev-protocol-springcloud-message-study/rocket-mq/message-with-key
|
||||
Content-Type: application/json
|
||||
|
||||
### message-with-tag
|
||||
GET http://127.0.0.1:8001/dev-protocol-springcloud-message-study/rocket-mq/message-with-tag
|
||||
Content-Type: application/json
|
||||
|
||||
### message-with-all
|
||||
GET http://127.0.0.1:8001/dev-protocol-springcloud-message-study/rocket-mq/message-with-all
|
||||
Content-Type: application/json
|
||||
@ -0,0 +1,7 @@
|
||||
### wrong-rollback-for
|
||||
GET http://127.0.0.1:8001/imooc-study-ecommerce-dev/transactional-lose/wrong-rollback-for
|
||||
Content-Type: application/json
|
||||
|
||||
### wrong-inner-call
|
||||
GET http://127.0.0.1:8001/imooc-study-ecommerce-dev/transactional-lose/wrong-inner-call
|
||||
Content-Type: application/json
|
||||
Loading…
Reference in New Issue