实战Kafka+springboot的经验记录

在当前微服务项目体系中,不同服务之间的通信功能十分重要,之前有一个项目选择了kafka做为技术选型.所以写个博客记录一下大概的工作业务步骤. 在实现的过程中,kafka的搭建基本上没有什么难度,关键在于不同的消息类型之间要进行不同的业务逻辑处理,在实际项目中使用了策略模式来解决这个问题
1.首先下载依赖,搭建配置文件,这些是直接上网查询就行,不用强行记录
2.新建一个接口类

public interface MyKafkaHandler {
    
    void registerUser(BaseMessage baseMessage);

    
    void payOrder(BaseMessage baseMessage);

    
    void processingPoints(BaseMessage baseMessage);

3. 新建一个策略模式工厂类

public class MyFactoryHashMap {

    private static HashMap handlerHashMap = new HashMap<>();

    
    public static void registerHandler(String type, MyKafkaHandler myKafkaHandler) {
        handlerHashMap.put(type, myKafkaHandler);
    }

    
    public static MyKafkaHandler getHandlerByType(String type) {
        return StrUtil.isNotEmpty(type) && handlerHashMap.containsKey(type) ? handlerHashMap.get(type) : null;
    }
}

4.定义几个接口的实现类 并实现上面定义的接口 和InitializingBean 类,并在afterPropertiesSet 方法中将当前类的 type存到策略模式的工厂中

@Slf4j
@Component
public class PayOrderHandler implements MyKafkaHandler, InitializingBean {
    
    @Override
    public void registerUser(BaseMessage baseMessage) {

    }

    
    @Override
    public void payOrder(BaseMessage baseMessage) {
        log.info("----进入接口payOrder------"+baseMessage.getId());
    }

    
    @Override
    public void processingPoints(BaseMessage baseMessage) {

    }

    @Override
    public void afterPropertiesSet() throws Exception {
        MyFactoryHashMap.registerHandler(MyKafkaTypeEnum.PAY_ORDER.name(), this);
    }

@Slf4j
@Component
public class ProcessingPointsHandler implements MyKafkaHandler, InitializingBean {
    
    @Override
    public void registerUser(BaseMessage baseMessage) {

    }

    
    @Override
    public void payOrder(BaseMessage baseMessage) {

    }

    
    @Override
    public void processingPoints(BaseMessage baseMessage) {
        log.info("-----处理积分------"+baseMessage.getId());
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        MyFactoryHashMap.registerHandler(MyKafkaTypeEnum.INTEGRAL.name(),this);
    }
}

@Slf4j
@Component
public class RegisterUserHandler implements MyKafkaHandler , InitializingBean {
    
    @Override
    public void registerUser(BaseMessage baseMessage) {
        log.info("-----接收消息处理注册---->"+baseMessage.getId());
    }

    
    @Override
    public void payOrder(BaseMessage baseMessage) {

    }

    
    @Override
    public void processingPoints(BaseMessage baseMessage) {

    }

    @Override
    public void afterPropertiesSet() throws Exception {
        MyFactoryHashMap.registerHandler(MyKafkaTypeEnum.REGISTER.name(),this);
    }
}

5. 其他需要的两个常量类可以自定义

@Getter
@AllArgsConstructor
public enum MyKafkaTypeEnum {

    REGISTER("注册"),
    PAY_ORDER("支付"),
    INTEGRAL("积分");

    private String description;


    
    public static  MyKafkaTypeEnum resultEnumByName(String name){
        if (StrUtil.isEmpty(name)) {
            return null;
        }
        for (MyKafkaTypeEnum value : values()) {
            if (value.name().equals(name)) {
                return value;
            }
        }
        return null;
    }

@AllArgsConstructor
@Data
public class BaseMessage {
    private String id;
}

6. 测试验证工厂模式是否生效,由于本地没有搭建kafka的环境.所以用controller模拟消息推送

    
    @GetMapping("/imitateKafkaMessage")
    public ResultVo imitateKafkaMessage(@RequestParam String id) {

        //模拟通过kafka发送消息 通过 BaseMessage ID获取 消息对象本身
        // 这里 假设获取 用户role ，其实逻辑是一样的 只是换一个不同的对象
        Role role = roleService.getById(id);
        if (Objects.nonNull(role)&& StrUtil.isNotEmpty(role.getRoleName())) {
            MyKafkaTypeEnum myKafkaTypeEnum = MyKafkaTypeEnum.resultEnumByName(role.getRoleName());
            if (Objects.nonNull(myKafkaTypeEnum)) {
                BaseMessage baseMessage = new BaseMessage(id);
                MyFactoryHashMap.getHandlerByType(myKafkaTypeEnum.name()).registerUser(baseMessage);
            }
        }

        return new ResultVo<>(StatusCodeEnum.SUCCESS_CODE.getCode(), "", null);
    }

当ID为4的时候,输出结果为

说明进入到了注册用户的具体实现类,当ID为5的时候,却不会打印任何数据
数据库的ID与对应的值