2消息订阅场景实现

1、背景

   随着微服务的发展，业务流程变得越发复杂，经常出现一个业务完成后需要通知多个下游系统或上游系统当前业务的状态。目前我们的业务代码中会有很多的业务通知方法，通知组装逻辑，让本该简单的业务变得复杂，难以维护。
   为了使通知逻辑变得职责单一，与主逻辑解耦，写了这个框架。

1.1、以前

    @Test
    public void test() {
        CReq cReq = new CReq();
        cReq.setName("张三");
        this.notifyC(cReq);

        DReq dReq = new DReq();
        dReq.setName("张三");
        this.notifyD(dReq);

        // 当然也有人会将通知的数据结构统一
    }

    private void notifyD(DReq dReq) {
        log.info("CObserver消费数据:{}", dReq.getName());
    }

    private void notifyC(CReq cReq) {
        log.info("DObserver消费数据:{}", cReq.getName());
    }

以前我们可能会在主逻辑中写出所有需要通知的逻辑，而不同的通知主体又会有不同的通知数据结构。

所以在住逻辑中会存在组装通知1、组装通知2，等，如果需要通知第3个，也会在当前主逻辑中添加。久而久之主逻辑下会有很大篇幅是用来组装通知逻辑的。如果需要扩展其他的通知也会变得复杂，使不同开发人员不知道在何处写通知逻辑，如何取数的问题，甚至可能写出影响主逻辑的代码（如容错）。

1.2、现在

@Autowired
    private CObserver cObserver;
    @Autowired
    private DObserver dObserver;

    @Test
    public void bizSubjectTest() {
        BizSubject bizSubject = new BizSubject();
        bizSubject.add(cObserver);
        bizSubject.add(dObserver);

        TestReq testReq = new TestReq();
        testReq.setName("张三");
        bizSubject.publish(testReq);
    }

1、可以做到消息通知数据结构统一，不用考虑取数不统一的问题。

2、不用考虑在何处添加通知逻辑。

3、不用考虑主逻辑的具体逻辑。

4、可以统一的对通知逻辑做容错处理，或其他附加操作。

2、类图

3、时序图

4、实现

4.1、鱼骨图

4.2、framework

4.2.1、BizSubject

• 主题，业务需要发布通知的主题

package com.guo.core.framework.observable;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

@Slf4j
public class BizSubject {

    private List<BizObserver> consumers = new ArrayList<>();

    public void add(BizObserver bizObserver) {
        consumers.add(bizObserver);
    }

    public <T> void publish(T data) {
        consumers.forEach(consumer -> {
            try {
                consumer.consume(data);
            } catch (Exception e) {
                log.error("通知异常", e);
            }
        });
    }

}

4.2.2、BizObserver

• 业务主题具体的通知实现

package com.guo.core.framework.observable;

public interface BizObserver<T> {

    void consume(T data);

}

4.3、implement

4.3.1、CObserver

• 模拟C观察者接收主题通知

package com.guo.core.framework.observable;

import com.guo.core.framework.template.TestReq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class CObserver implements BizObserver<TestReq>{
    @Override
    public void consume(TestReq data) {
      log.info("CObserver消费数据:{}", data.getName());
    }
}

4.3.2、DObserver

• 模拟D观察者接收主题通知

package com.guo.core.framework.observable;

import com.guo.core.framework.template.TestReq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class DObserver implements BizObserver<TestReq>{
    @Override
    public void consume(TestReq data) {
        log.info("DObserver消费数据:{}", data.getName());
    }
}

4.4、client

• 集成测试

package com.guo.core.framework.observable;

import com.guo.core.framework.ThirdServiceApplication;
import com.guo.core.framework.template.TestReq;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ThirdServiceApplication.class)
public class ObservableTest {

    @Test
    public void test() {
        CReq cReq = new CReq();
        cReq.setName("张三");
        this.notifyC(cReq);

        DReq dReq = new DReq();
        dReq.setName("张三");
        this.notifyD(dReq);

        // 当然也有人会将通知的数据结构统一
    }

    private void notifyD(DReq dReq) {
        log.info("CObserver消费数据:{}", dReq.getName());
    }

    private void notifyC(CReq cReq) {
        log.info("DObserver消费数据:{}", cReq.getName());
    }

    @Autowired
    private CObserver cObserver;
    @Autowired
    private DObserver dObserver;

    @Test
    public void bizSubjectTest() {
        BizSubject bizSubject = new BizSubject();
        bizSubject.add(cObserver);
        bizSubject.add(dObserver);

        TestReq testReq = new TestReq();
        testReq.setName("张三");
        bizSubject.publish(testReq);
    }

}