public interface Breakfast {
    void eat();
}

public enum BreakfastEnum implements Breakfast {


    Beijing("北京") {
        @Override
        public void eat() {
            System.out.println("北京人早餐吃豆汁和焦圈");
        }
    },

    Wuhan("武漢") {
        @Override
        public void eat() {
            System.out.println("武漢人早餐吃熱干面、豆皮...");
        }
    },

    Unknown("未知") {
        @Override
        public void eat() {
            System.out.println("不吃早餐！");
        }
    };

    private String city;

    BreakfastEnum(String city) {
        this.city = city;
    }

    private String getCity() {
        return this.city;
    }

    /**
     * 提供統(tǒng)一入口 找到對應(yīng)子類并執(zhí)行
     *
     * @param city
     */
    public static void eat(String city) {
        BreakfastEnum[] values = BreakfastEnum.values();
        Arrays.stream(values).filter(e -> city.equals(e.city)).findFirst().orElse(Unknown).eat();
    }
}

public interface MessageHandle {

    /**
     * 發(fā)布消息
     *
     * @param msg
     */
    void publish(JSONObject msg);
}

/**
 * 發(fā)送短信
 */
@Service("sms")
public class SmsMessageHandle implements MessageHandle {
    @Override
    public void publish(JSONObject msg) {
        // 發(fā)送短信
        // 省略實現(xiàn)...
    }

}

/**
 * 推送
 */
@Service("push")
public class PushMessageHandle implements MessageHandle {
    @Override
    public void publish(JSONObject msg) {
        // 推送到app
        // 省略實現(xiàn)...
    }

}

// 1.指定途徑發(fā)送
@Component
public class MessageListener {

    @Autowired
    private Map<String, MessageHandle> messageHandleMap;
    
    @KafkaListener(groupId = "message-server", topics = "message")
    public void listener(String message, Acknowledgment ack) {
        JSONObject messageJson = JSON.parseObject(message);
        // 獲取發(fā)布方式 sms push...對應(yīng)實現(xiàn)@Service注解中的名稱
        String publishType =  messageJson.getString("publishType");
        // 獲取實現(xiàn)
        MessageHandle handle = messageHandleMap.get(publishType);
        if(handle != null) {
            handle.publish(messageJson);
        }
        // 提交偏移量
        ack.acknowledge();
    }
    
 }
    

// 2.每種途徑都發(fā)送
@Component
public class MessageListener {

    @Autowired
    private List<MessageHandle> messageHandleList;
    
    @KafkaListener(groupId = "message-server", topics = "message")
    public void listener(String message, Acknowledgment ack) {
        JSONObject messageJson = JSON.parseObject(message);
        // 每種途徑都發(fā)送
        for(MessageHandle handle : messageHandleList){
            handle.publish(messageJson);
        }
        // 提交偏移量
        ack.acknowledge();
    }
    
 }