Learning Spring Cloud

// Register as a Bean and enable load balancing
@Bean
@LoadBalanced
public RestTemplate restTemplate() {
    return new RestTemplate();
}

// Call style (service name replaces IP:Port)
String url = "http://user-service/api/users/" + userId;
User user = restTemplate.getForObject(url, User.class);

docker run -d \
  --name nacos \
  -e MODE=standalone \
  -p 8848:8848 \
  nacos/nacos-server:v2.1.0

<!-- Spring Cloud Alibaba dependency management -->
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-alibaba-dependencies</artifactId>
            <version>2021.0.5.0</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<!-- Nacos service-discovery dependency -->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>

spring:
  application:
    name: user-service        # service name, the identifier registered to Nacos
  cloud:
    nacos:
      server-addr: localhost:8848
      discovery:
        namespace: dev        # namespace ID (not the name)
        group: DEFAULT_GROUP

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>

@SpringBootApplication
@EnableFeignClients
public class OrderApplication {
    public static void main(String[] args) {
        SpringApplication.run(OrderApplication.class, args);
    }
}

@FeignClient(value = "user-service")   // corresponds to spring.application.name
public interface UserClient {

    @GetMapping("/api/users/{id}")
    User getUserById(@PathVariable("id") Long id);

    @PostMapping("/api/users")
    User createUser(@RequestBody UserDTO dto);
}

@Service
@RequiredArgsConstructor
public class OrderService {
    private final UserClient userClient;

    public OrderVO getOrder(Long orderId) {
        Order order = orderMapper.selectById(orderId);
        User user = userClient.getUserById(order.getUserId()); // like calling a local method
        return buildVO(order, user);
    }
}

<dependency>
    <groupId>io.github.openfeign</groupId>
    <artifactId>feign-okhttp</artifactId>
</dependency>

feign:
  okhttp:
    enabled: true
  httpclient:
    ok-http:
      connection-pool-timeout: PT0.0005S
      max-connections: 200
      max-connections-per-route: 50

<dependency>
    <groupId>com.hmall</groupId>
    <artifactId>hm-api</artifactId>
    <version>${project.version}</version>
</dependency>

// Way 1: specify the scan package path
@EnableFeignClients(basePackages = "com.hmall.api.client")

// Way 2: specify the FeignClient classes (more precise)
@EnableFeignClients(clients = {UserClient.class, ItemClient.class})

// Global config Bean
@Bean
public Logger.Level feignLogLevel() {
    return Logger.Level.FULL;
}

# Enable logging for a specific FeignClient (with logging.level)
logging:
  level:
    com.hmall.api.client.UserClient: DEBUG

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-gateway</artifactId>
</dependency>
<!-- The gateway also needs to register to Nacos to route via service discovery -->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>

spring:
  cloud:
    gateway:
      routes:
        - id: user-route                          # route ID, unique
          uri: lb://user-service                  # lb:// means load-balance to the service in Nacos
          predicates:
            - Path=/api/users/**                  # path predicate
          filters:
            - StripPrefix=1                       # strip the path prefix
        - id: order-route
          uri: lb://order-service
          predicates:
            - Path=/api/orders/**
            - Method=GET,POST                     # method predicate
            - Header=X-Request-Id, \d+            # header predicate (regex)
          filters:
            - AddRequestHeader=X-Source, gateway  # add a request header

@Component
@RequiredArgsConstructor
@Order(-1)   // priority; the smaller, the earlier it runs
public class AuthGlobalFilter implements GlobalFilter {

    private final JwtTool jwtTool;

    // Whitelist paths (no auth needed)
    private static final List<String> WHITE_LIST = List.of(
        "/api/users/login",
        "/api/users/register"
    );

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        String path = request.getPath().value();

        // 1. Let whitelisted paths through
        if (isWhitePath(path)) {
            return chain.filter(exchange);
        }

        // 2. Get the token
        String token = getTokenFromRequest(request);
        if (token == null) {
            return unauthorized(exchange);
        }

        // 3. Parse the token
        Long userId;
        try {
            userId = jwtTool.parseToken(token);
        } catch (Exception e) {
            return unauthorized(exchange);
        }

        // 4. Pass the user ID to downstream microservices
        ServerWebExchange mutatedExchange = exchange.mutate()
            .request(builder -> builder.header("X-User-Id", userId.toString()))
            .build();

        return chain.filter(mutatedExchange);
    }

    private boolean isWhitePath(String path) {
        return WHITE_LIST.stream().anyMatch(path::startsWith);
    }

    private String getTokenFromRequest(ServerHttpRequest request) {
        List<String> headers = request.getHeaders().get("Authorization");
        if (headers == null || headers.isEmpty()) return null;
        String auth = headers.get(0);
        return auth.startsWith("Bearer ") ? auth.substring(7) : null;
    }

    private Mono<Void> unauthorized(ServerWebExchange exchange) {
        exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
        return exchange.getResponse().setComplete();
    }
}

public class UserInfoInterceptor implements HandlerInterceptor {

    @Override
    public boolean preHandle(HttpServletRequest request,
                             HttpServletResponse response, Object handler) {
        String userIdStr = request.getHeader("X-User-Id");
        if (StringUtils.hasText(userIdStr)) {
            UserContext.setUser(Long.parseLong(userIdStr));
        }
        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response,
                                Object handler, Exception ex) {
        UserContext.removeUser(); // prevent memory leaks!
    }
}

// ThreadLocal utility class
public class UserContext {
    private static final ThreadLocal<Long> TL = new ThreadLocal<>();
    public static void setUser(Long userId) { TL.set(userId); }
    public static Long getUser() { return TL.get(); }
    public static void removeUser() { TL.remove(); }
}

// common module: resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
com.hmall.common.config.MvcConfig

# Exclude it in the gateway module's application.yml
spring:
  autoconfigure:
    exclude:
      - com.hmall.common.config.MvcConfig

@Bean
public RequestInterceptor userInfoRequestInterceptor() {
    return template -> {
        Long userId = UserContext.getUser();
        if (userId != null) {
            template.header("X-User-Id", userId.toString());
        }
    };
}

<!-- Nacos config center -->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
</dependency>
<!-- Bootstrap context, to load Nacos config before the app starts -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-bootstrap</artifactId>
</dependency>

# bootstrap.yml — pull shared config from Nacos
spring:
  application:
    name: user-service
  profiles:
    active: dev
  cloud:
    nacos:
      server-addr: localhost:8848
      config:
        file-extension: yaml
        shared-configs:
          - data-id: shared-jdbc.yaml      # shared database config
            group: DEFAULT_GROUP
            refresh: true
          - data-id: shared-redis.yaml     # shared Redis config
            group: DEFAULT_GROUP
            refresh: true
        # the service's own config: user-service-dev.yaml (auto-recognized)

// Way 1: @RefreshScope (refresh the whole Bean)
@Component
@RefreshScope
public class CartProperties {
    @Value("${cart.max-amount:100}")
    private Integer maxAmount;
}

// Way 2: @ConfigurationProperties (recommended, safer)
@Component
@ConfigurationProperties(prefix = "cart")
@Data
public class CartProperties {
    private Integer maxAmount = 100;
    // Auto-refreshes after a change in Nacos, no @RefreshScope needed
}

@Component
@RequiredArgsConstructor
public class DynamicRouteLoader implements ApplicationRunner {

    private final RouteDefinitionWriter writer;
    private final NacosConfigManager nacosConfigManager;

    private static final String DATA_ID = "gateway-routes.json";
    private static final String GROUP = "DEFAULT_GROUP";

    @Override
    public void run(ApplicationArguments args) throws Exception {
        // Initial pull
        loadRoutes();
        // Listen for changes
        nacosConfigManager.getConfigService().addListener(DATA_ID, GROUP,
            new Listener() {
                @Override
                public void receiveConfigInfo(String configInfo) {
                    loadRoutes();
                }
                @Override
                public Executor getExecutor() { return null; }
            });
    }

    private void loadRoutes() {
        // Parse JSON and register route definitions
        // ...
    }
}

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
</dependency>

spring:
  cloud:
    sentinel:
      transport:
        dashboard: localhost:8090   # Sentinel console address
      eager: true                  # initialize early, so the first request's resource is visible

java -Dserver.port=8090 \
     -Dcsp.sentinel.dashboard.server=localhost:8090 \
     -Dproject.name=sentinel-dashboard \
     -jar sentinel-dashboard.jar

feign:
  sentinel:
    enabled: true

// 1. Implement FallbackFactory
@Component
public class UserClientFallbackFactory implements FallbackFactory<UserClient> {

    @Override
    public UserClient create(Throwable cause) {
        log.error("UserClient call failed", cause);
        return new UserClient() {
            @Override
            public User getUserById(Long id) {
                // Return a default or empty object to avoid NPE
                return User.builder().id(id).name("Unknown user").build();
            }
        };
    }
}

// 2. Reference the factory in the FeignClient
@FeignClient(value = "user-service", fallbackFactory = UserClientFallbackFactory.class)
public interface UserClient {
    @GetMapping("/api/users/{id}")
    User getUserById(@PathVariable("id") Long id);
}

// 3. Register the factory as a Bean (if in the hm-api module, ensure it's scanned)
@Bean
public UserClientFallbackFactory userClientFallbackFactory() {
    return new UserClientFallbackFactory();
}

server:
  port: 8082
  tomcat:
    threads:
      max: 25           # max worker threads
      min-spare: 5      # min idle threads
    accept-count: 25    # wait-queue size
    max-connections: 100

CREATE DATABASE IF NOT EXISTS `seata`;
USE `seata`;

CREATE TABLE IF NOT EXISTS `global_table` (
    `xid` VARCHAR(128) NOT NULL,
    `transaction_id` BIGINT,
    `status` TINYINT NOT NULL,
    `application_id` VARCHAR(32),
    `transaction_service_group` VARCHAR(32),
    `transaction_name` VARCHAR(128),
    `timeout` INT,
    `begin_time` BIGINT,
    `application_data` VARCHAR(2000),
    `gmt_create` DATETIME,
    `gmt_modified` DATETIME,
    PRIMARY KEY (`xid`)
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4;

-- branch_table, lock_table, distributed_lock are similar (see the official docs)

docker run --name seata \
  -p 8099:8099 \
  -p 7099:7099 \
  -e SEATA_IP=192.168.184.129 \
  -v ./seata:/seata-server/resources \
  --privileged=true \
  --network testNet \
  -d seataio/seata-server:1.5.2

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
</dependency>

seata:
  registry:
    type: nacos
    nacos:
      server-addr: localhost:8848
      namespace: ""
      group: DEFAULT_GROUP
      application: seata-server   # the Seata Server's service name in Nacos
  tx-service-group: hmall         # transaction group name
  service:
    vgroup-mapping:
      hmall: default              # transaction group → cluster mapping
  data-source-proxy-mode: AT      # AT mode (default)

@Service
@RequiredArgsConstructor
public class OrderService {

    @GlobalTransactional(name = "create-order", rollbackFor = Exception.class)
    public void createOrder(OrderDTO dto) {
        // 1. Deduct stock (remote call → Seata RM registers a branch transaction)
        itemClient.deductStock(dto.getItemId(), dto.getNum());
        // 2. Deduct balance (remote call → Seata RM registers a branch transaction)
        userClient.deductBalance(dto.getUserId(), dto.getAmount());
        // 3. Create the order (local operation → Seata RM registers a branch transaction)
        save(buildOrder(dto));
    }
}

seata:
  data-source-proxy-mode: XA

// Enable XA support
@Bean
@ConfigurationProperties(prefix = "spring.datasource")
public DataSource dataSource(DruidDataSourceWrapper druidDataSourceWrapper) {
    return new DataSourceProxyXA(druidDataSourceWrapper);
}

CREATE TABLE IF NOT EXISTS `undo_log` (
    `branch_id`     BIGINT NOT NULL COMMENT 'branch transaction id',
    `xid`           VARCHAR(128) NOT NULL COMMENT 'global transaction id',
    `context`       VARCHAR(128) NOT NULL COMMENT 'serialization',
    `rollback_info` LONGBLOB NOT NULL COMMENT 'rollback info',
    `log_status`    INT(11) NOT NULL COMMENT '0:normal,1:defense',
    `log_created`   DATETIME(6) NOT NULL,
    `log_modified`  DATETIME(6) NOT NULL,
    UNIQUE KEY `ux_undo_log` (`xid`, `branch_id`)
) ENGINE = InnoDB AUTO_INCREMENT = 1 DEFAULT CHARSET = utf8mb4;

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring:
  rabbitmq:
    host: localhost
    port: 5672
    virtual-host: /hmall
    username: hmall
    password: 123456

@Service
@RequiredArgsConstructor
public class PayService {
    private final RabbitTemplate rabbitTemplate;

    public void notifyPaySuccess(Long orderId) {
        rabbitTemplate.convertAndSend(
            "pay.direct",     // exchange
            "pay.success",    // routing key
            orderId           // message body (auto-serialized when using the JSON converter)
        );
    }
}

@Component
@Slf4j
public class PayMessageListener {

    @RabbitListener(queues = "pay.queue")
    public void onPaySuccess(Long orderId) {
        log.info("received pay-success message, order ID: {}", orderId);
        // update the order status
    }
}

// Declaration (consumer side)
@Bean
public FanoutExchange fanoutExchange() {
    return new FanoutExchange("hmall.fanout");
}

@Bean
public Queue fanoutQueue1() {
    return QueueBuilder.durable("fanout.queue1").build();
}

@Bean
public Binding binding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
    return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
}

// Annotation-based declaration (more concise, recommended)
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue1", durable = "true"),
    exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
    key = {"red", "blue"}
))
public void listenDirectQueue1(String message) {
    log.info("direct.queue1 received: {}", message);
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue1", durable = "true"),
    exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),
    key = "china.#"   // matches all routing keys starting with china.
))
public void listenTopicQueue1(String message) {
    log.info("topic.queue1 received: {}", message);
}

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
</dependency>

// Both publisher and consumer need this config
@Bean
public MessageConverter jacksonMessageConverter() {
    Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
    converter.setCreateMessageIds(true); // auto-generate message IDs for idempotency checks
    return converter;
}

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1   # deliver at most 1 message at a time; the next only after this one is processed

spring:
  rabbitmq:
    publisher-confirm-type: correlated  # async callback confirmation (recommended)
    publisher-returns: true             # enable routing-failure callback
    template:
      mandatory: true

@PostConstruct  // runs after the Bean is initialized
public void init() {
    // Callback on routing failure (message reached the Exchange but wasn't routed to a Queue)
    rabbitTemplate.setReturnsCallback(returnedMessage -> {
        log.error("message routing failed: exchange={}, routingKey={}, message={}",
            returnedMessage.getExchange(),
            returnedMessage.getRoutingKey(),
            returnedMessage.getMessage());
        // resend or log it
    });
}

public void sendWithConfirm(String exchange, String key, Object msg) {
    CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
    cd.getFuture().addCallback(
        confirm -> {
            if (confirm.isAck()) {
                log.info("message reached the Exchange, ID: {}", cd.getId());
            } else {
                log.error("message didn't reach the Exchange, reason: {}, ID: {}", confirm.getReason(), cd.getId());
                // resend logic
            }
        },
        throwable -> log.error("message send exception", throwable)
    );
    rabbitTemplate.convertAndSend(exchange, key, msg, cd);
}

// Exchange persistence (durable=true by default)
@Bean
public DirectExchange payExchange() {
    return ExchangeBuilder.directExchange("pay.direct").durable(true).build();
}

// Queue persistence
@Bean
public Queue payQueue() {
    return QueueBuilder.durable("pay.queue").build();
}

// Message persistence (Spring AMQP defaults to persistent)
rabbitTemplate.convertAndSend(exchange, key, msg, message -> {
    message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
    return message;
});

// Bean way
@Bean
public Queue lazyQueue() {
    return QueueBuilder.durable("lazy.queue")
        .lazy()   // set as a lazy queue
        .build();
}

// Annotation way
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(
        name = "lazy.queue",
        durable = "true",
        arguments = @Argument(name = "x-queue-mode", value = "lazy")
    ),
    ...
))

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto  # recommended: around-enhancement, auto ack/nack
        # none: ack immediately, unsafe
        # manual: manual ack, flexible but intrusive to code

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true
          initial-interval: 1000ms  # initial retry interval
          multiplier: 1.0           # interval multiplier
          max-attempts: 3           # max retries
          stateless: true           # stateless retry

@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate) {
    // Recommended: after retries are exhausted, deliver the failed message to a designated "error exchange"
    return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}

@RabbitListener(queues = "pay.queue")
public void onPaySuccess(Message message) {
    String msgId = message.getMessageProperties().getMessageId();

    // Way 1: dedupe with message ID + Redis
    Boolean isFirst = redisTemplate.opsForValue()
        .setIfAbsent("pay:msg:" + msgId, "1", 30, TimeUnit.MINUTES);
    if (Boolean.FALSE.equals(isFirst)) {
        log.warn("duplicate message, ignoring: {}", msgId);
        return;
    }

    Long orderId = (Long) rabbitTemplate.getMessageConverter().fromMessage(message);

    // Way 2: check the business state
    Order order = orderService.getById(orderId);
    if (order.getStatus() != OrderStatus.UNPAID) {
        log.warn("order status already updated, ignoring duplicate message: {}", orderId);
        return;
    }

    orderService.updateStatus(orderId, OrderStatus.PAID);
}

@Bean
public Queue ttlQueue() {
    return QueueBuilder.durable("ttl.queue")
        .ttl(30000)                          // message lives 30 seconds
        .deadLetterExchange("dlx.direct")    // forwards to the DLX after expiry
        .deadLetterRoutingKey("dlx.order")
        .build();
}

@Bean
public DirectExchange delayedExchange() {
    return ExchangeBuilder.directExchange("delayed.exchange")
        .delayed()   // declare as a delayed exchange
        .durable(true)
        .build();
}

// Specify the delay when sending
rabbitTemplate.convertAndSend("delayed.exchange", "delay.key", orderId, message -> {
    message.getMessageProperties().setDelayLong(30 * 60 * 1000L); // delay 30 minutes
    return message;
});