Java开发项目的中间件介绍

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Service;
import java.util.concurrent.TimeUnit;

@Service
public class RedisService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate; // 可存储任意Java对象
    // 也可注入 StringRedisTemplate，专门用于操作字符串

    /**
     * 设置键值对（带过期时间）
     * @param key 键
     * @param value 值
     * @param expire 过期时间（秒）
     */
    public void set(String key, Object value, long expire) {
        ValueOperations<String, Object> ops = redisTemplate.opsForValue();
        ops.set(key, value, expire, TimeUnit.SECONDS);
    }

    /**
     * 获取值
     * @param key 键
     * @return 值
     */
    public Object get(String key) {
        ValueOperations<String, Object> ops = redisTemplate.opsForValue();
        return ops.get(key);
    }

    /**
     * 删除键
     * @param key 键
     * @return 是否删除成功
     */
    public Boolean delete(String key) {
        return redisTemplate.delete(key);
    }
}

@SpringBootApplication
@EnableCaching // 开启缓存注解功能
public class MyApplication {
    public static void main(String[] args) {
        SpringApplication.run(MyApplication.class, args);
    }
}

@Service
public class UserService {

    @Cacheable(value = "user", key = "#id") // 如果缓存中有，直接返回；没有则执行方法并存入缓存
    public User getUserById(Long id) {
        // 模拟从数据库查询
        System.out.println("从数据库查询用户: " + id);
        return new User(id, "User" + id);
    }

    @CacheEvict(value = "user", key = "#id") // 删除缓存
    public void deleteUserById(Long id) {
        // 先删除数据库数据，此注解会自动清除缓存
    }

    @CachePut(value = "user", key = "#user.id") // 更新缓存
    public User updateUser(User user) {
        // 更新数据库...
        return user; // 返回的结果会被缓存
    }
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.RedisCacheConfiguration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import java.time.Duration;

@Configuration
public class RedisCacheConfig {

    @Bean
    public RedisCacheManager cacheManager(RedisConnectionFactory connectionFactory) {
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
                .entryTtl(Duration.ofMinutes(10)) // 设置默认过期时间10分钟
                .disableCachingNullValues(); // 不缓存null值

        return RedisCacheManager.builder(connectionFactory)
                .cacheDefaults(config)
                .build();
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Service
public class KafkaProducerService {

    private static final String TOPIC_NAME = "my_topic";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 发送消息
     * @param message 消息内容
     */
    public void sendMessage(String message) {
        // 发送消息，topic不存在时会自动创建
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(TOPIC_NAME, message);

        // 添加回调，处理发送成功或失败后的逻辑
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println("Sent message: [" + message + "] with offset: [" + result.getRecordMetadata().offset() + "]");
            }

            @Override
            public void onFailure(Throwable ex) {
                System.err.println("Unable to send message: [" + message + "] due to: " + ex.getMessage());
            }
        });
    }
}

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    // 监听名为'my_topic'的Topic，concurrency表示并发消费者线程数
    @KafkaListener(topics = "my_topic", groupId = "my-group", concurrency = "3")
    public void listen(@Payload String message, Acknowledgment ack) {
        try {
            System.out.println("Received Message: " + message);
            // 模拟业务处理...
            // 处理成功后，手动提交偏移量
            ack.acknowledge();
        } catch (Exception e) {
            // 处理异常，可根据业务决定是重试还是记录日志等
            System.err.println("Error processing message: " + message);
            // 不确认消息，Kafka会根据配置进行重试
        }
    }
}

// 生产者
kafkaTemplate.send(TOPIC_NAME, new MyObject(1L, "Name"));

// 消费者
@KafkaListener(topics = "my_topic", groupId = "my-group")
public void listen(MyObject myObject, Acknowledgment ack) {
    System.out.println("Received Object: " + myObject);
    ack.acknowledge();
}