基于【觀察者設(shè)計模式】設(shè)計異步多渠道群發(fā)框架

前言:設(shè)計模式源于生活

觀察者基本概念

觀察者模式，又可以稱之為發(fā)布-訂閱模式，觀察者，顧名思義，就是一個監(jiān)聽者，類似監(jiān)聽器的存在，一旦被觀察/監(jiān)聽的目標發(fā)生的情況，就會被監(jiān)聽者發(fā)現(xiàn)，這么想來目標發(fā)生情況到觀察者知道情況，其實是由目標將情況發(fā)送到觀察者的
白話文:當一個對象發(fā)生改變的時候,可以通知其他所有對象

概念很清晰,舉個栗子來理解一下觀察者模式的含義,我們都在抖音關(guān)注了某位大咖的時候,每當這位大咖更新了一條動態(tài)時候,關(guān)注大咖的粉絲都能收到通知,簡單用一張圖來表明他們之間的關(guān)系

上面這位大咖發(fā)布了個動態(tài),然后他的粉絲都收到通知,并知曉了,從這個栗子可以看到,這里包含兩類人,一是大咖，二是粉絲,那么翻譯到程序中語言就是觀察者的主題和觀察者

那么大咖就相當于主題，粉絲相當于觀察者,隨時觀察大咖的動態(tài)消息,不過大咖也有權(quán)力拉黑你或者讓你關(guān)注,那么從類圖的角度了解一下

observer:抽象觀察者，是觀察者者的抽象類，它定義了一個更新接口，使得在得到主題更改通知時更新自己。這就是我們所有粉絲的抽象
ConcrereObserver：具體觀察者，實現(xiàn)抽象觀察者定義的更新接口，以便在得到主題更改通知時更新自身的狀態(tài)。具體每一個粉絲
Subject：抽象主題，他把所有觀察者對象保存在一個集合里，可以有任意數(shù)量的觀察者，抽象主題提供一個接口，可以增加和刪除觀察者對象。意思就是大咖把所有的粉絲都保存在一個賬號里面，粉絲數(shù)量不限，可以新增粉絲也可以拉黑粉絲
ConcreteSubject：具體主題，該角色將有關(guān)狀態(tài)存入具體觀察者對象，在具體主題的內(nèi)部狀態(tài)發(fā)生改變時，給所有注冊過的觀察者發(fā)送通知。意思是我們的大咖一有動態(tài)，就會把消息給粉絲。

觀察者應(yīng)用場景

1.對一個對象狀態(tài)的更新，需要其他對象同步更新，而且其他對象的數(shù)量動態(tài)可變
2.對象僅需要將自己的更新通知給其他對象而不需要知道其他對象的細節(jié)
例如:
1.分布式配置中心,當配置發(fā)生改變,通過事件監(jiān)聽來刷新配置,常見的有,apollo,nacos,spring config
2.zk的節(jié)點,當節(jié)點發(fā)生變化,會通知所有的客戶端
3.多渠道群發(fā),當你關(guān)注了某位大咖,那么每次更新動態(tài)的時候,所有關(guān)注大咖的人,都會收到大咖的動態(tài)更新消息提示

我這里通過三種方式,來實現(xiàn)異步多渠道群發(fā)框架

三種形式的maven依賴都是一樣的

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.1.RELEASE</version>
    </parent>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>29.0-jre</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.12</version>
            <!--            <scope>provided</scope>-->
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>

    </dependencies>

第一種方式,基于java內(nèi)存的形式實現(xiàn)

抽象觀察者

public interface ObServer {

    /**
     * 發(fā)送消息
     *
     * @param msg
     */
    void sendMsg(String msg);
}

短信具體觀察者

@Slf4j
public class SmsObServer implements ObServer {

    @Override
    public void sendMsg(String msg) {
        log.info("發(fā)送短信消息,內(nèi)容:{}", msg);
    }
}

郵件具體觀察者

@Slf4j
public class EmailObServer implements ObServer {

    @Override
    public void sendMsg(String msg) {
        log.info("發(fā)送郵件消息,內(nèi)容:{}", msg);
    }
}

抽象主題

@Slf4j
public abstract class SunnySubject {

    protected List<ObServer> obServerList = Lists.newArrayList();

    /**
     * 注冊觀察者
     *
     * @param obServer
     */
    public void addObServer(ObServer obServer) {
        obServerList.add(obServer);
    }

    /**
     * 移除觀察者
     *
     * @param obServer
     */
    public void removeObServer(ObServer obServer) {
        boolean contains = obServerList.contains(obServer);
        if (contains) {
            obServerList.remove(obServer);
        }
    }

    /**
     * 通知觀察者
     */
    public abstract void notifyObServer(String msg);
}

具體主題

@Slf4j
public class ConcreteSubject extends SunnySubject {

    private ExecutorService executorService;

    public ConcreteSubject() {
        executorService = Executors.newFixedThreadPool(10);
    }

    @Override
    public void notifyObServer(String msg) {
        log.info("目標對象狀態(tài)已變化......發(fā)送通知給觀察者中");
        for (ObServer ob : obServerList) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    ob.sendMsg(msg);
                }
            });
        }
    }
}

啟動

public class Test {
    public static void main(String[] args) {
        //初始化哦
        SunnySubject sunnySubject = new ConcreteSubject();

        //注冊
        sunnySubject.addObServer(new SmsObServer());
        sunnySubject.addObServer(new EmailObServer());

        //通知
        sunnySubject.notifyObServer("你好,觀察者");
    }
}

第二種方式,基于SpringIOC容器形式實現(xiàn)

抽象觀察者

public interface ObServer {

    /**
     * 發(fā)送消息
     *
     * @param msg
     */
    void sendMsg(String msg);
}

短信具體觀察者

@Component
@Slf4j
public class SmsObServer implements ObServer {

    @Override
    public void sendMsg(String msg) {
        log.info("發(fā)送短信消息,內(nèi)容:{}", msg);
    }
}

郵件具體觀察者

@Component
@Slf4j
public class EmailObServer implements ObServer {

    @Override
    public void sendMsg(String msg) {
        log.info("發(fā)送郵件消息,內(nèi)容:{}", msg);
    }
}

抽象主題

@Slf4j
public abstract class SunnySubject {

    protected List<ObServer> obServerList = Lists.newArrayList();

    /**
     * 注冊觀察者
     *
     * @param obServer
     */
    public void addObServer(ObServer obServer) {
        obServerList.add(obServer);
    }

    /**
     * 移除觀察者
     *
     * @param obServer
     */
    public void removeObServer(ObServer obServer) {
        boolean contains = obServerList.contains(obServer);
        if (contains) {
            obServerList.remove(obServer);
        }
    }

    /**
     * 通知觀察者
     */
    public abstract void notifyObServer(String msg);
}

具體主題

@Component
@Slf4j
public class ConcreteSubject extends SunnySubject {

    private ExecutorService executorService;

    public ConcreteSubject() {
        executorService = Executors.newFixedThreadPool(10);
    }

    @Override
    public void notifyObServer(String msg) {
        log.info("目標對象狀態(tài)已變化......發(fā)送通知給觀察者中");
        for (ObServer ob : obServerList) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    ob.sendMsg(msg);
                }
            });
        }
    }
}

觀察者配置類

@Component
@Slf4j
public class ObServerConfig implements ApplicationRunner {

    @Autowired
    private SmsObServer smsObServer;

    @Autowired
    private EmailObServer emailObServer;

    @Autowired
    private ConcreteSubject concreteSubject;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        concreteSubject.addObServer(smsObServer);
        concreteSubject.addObServer(emailObServer);
    }
}

入口

@RestController
public class ObServerController {

    @Autowired
    private ConcreteSubject concreteSubject;

    @GetMapping("/send")
    public void test(){
        concreteSubject.notifyObServer("你好,觀察者");
    }

}

訪問:http://localhost:8080/send

當然,如上面我們的代碼還可以在優(yōu)化一下,實現(xiàn)動態(tài)注冊

@Component
@Slf4j
public class ObServerConfig implements ApplicationRunner, ApplicationContextAware {

    @Autowired
    private SmsObServer smsObServer;

    @Autowired
    private EmailObServer emailObServer;

    @Autowired
    private ConcreteSubject concreteSubject;

    private ApplicationContext applicationContext;

    @Override
    public void run(ApplicationArguments args) throws Exception {
//        concreteSubject.addObServer(smsObServer);
//        concreteSubject.addObServer(emailObServer);

        Map<String, ObServer> map = applicationContext.getBeansOfType(ObServer.class);
        for(String key : map.keySet()){
            ObServer observer = map.get(key);
            concreteSubject.addObServer(observer);
        }
    }

    /**
     * 獲取上下文環(huán)境對象得到Spring容器中的Bean
     *
     * @param applicationContext
     * @throws BeansException
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}

第三種方式,基于Spring事件形式實現(xiàn)

實體類

public class UserMessageEntity extends ApplicationEvent {

    private String email;

    private Long phone;

    private Long userId;

    public UserMessageEntity(Object source) {
        super(source);
    }

    public UserMessageEntity(Object source, String email, Long phone) {
        super(source);
        this.email = email;
        this.phone = phone;
    }

    @Override
    public String toString() {
        return "UserMessageEntity{" +
                "email='" + email + '\'' +
                ", phone=" + phone +
                ", userId=" + userId +
                '}';
    }
}

郵箱事件回調(diào)通知

@Component
@Slf4j
public class EmailListener implements ApplicationListener<UserMessageEntity> {

    @Override
    @Async
    public void onApplicationEvent(UserMessageEntity userMessageEntity) {
        log.info("郵箱通知:{}", userMessageEntity.toString());
    }
}

短信事件回調(diào)通知

@Component
@Slf4j
public class SmsListener implements ApplicationListener<UserMessageEntity> {

    @Override
    public void onApplicationEvent(UserMessageEntity userMessageEntity) {
        log.info("短信通知:{}", userMessageEntity.toString());
    }
}

入口

@RestController
public class ObServerController {

    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;

    @GetMapping("/send")
    public void sendTwo() {
        UserMessageEntity userMessageEntity = new UserMessageEntity(this, "123456@163.com", 15096111111L);
        applicationEventPublisher.publishEvent(userMessageEntity);
    }
}

訪問:http://localhost:8080/send

總結(jié)

觀察者模式的主要優(yōu)點在于可以實現(xiàn)表示層和數(shù)據(jù)邏輯層的分離，并在觀察目標和觀察者之間建立一個抽象的耦合，支持廣播通信；其主要缺點在于如果一個觀察目標對象有很多直接和間接的觀察者的話，將所有的觀察者都通知到會花費很多時間，而且如果在觀察者和觀察目標之間有循環(huán)依賴的話，觀察目標會觸發(fā)它們之間進行循環(huán)調(diào)用，可能導(dǎo)致系統(tǒng)崩潰。
其實還有一點需要我們?nèi)チ私?，在上面的例子當中我們的會發(fā)現(xiàn)，其實粉絲的消息是大咖推過來的，還有一種觀察者模式，也就是我們的粉絲主動去獲取消息。
（1）推模型： 主題對象向觀察者推送主題的詳細信息，不管是否需要。
（2）拉模型：主題對象在通知觀察者的時候，只傳遞少量信息。如果觀察者需要更具體的信息，由觀察者主動到主題對象中獲取

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

友情鏈接更多精彩內(nèi)容