1.发布者接口

package com.shoshana.publishsubscribe;

public interface IPublisher {

public void publish(SubscribePublish subscribePublish, M message, boolean isInstantMsg);

}

2.订阅者接口

package com.shoshana.publishsubscribe;

public interface ISubcriber {

public void subcribe(SubscribePublish subscribePublish);

public void unSubcribe(SubscribePublish subscribePublish);

public void update(String publisher, M message);

}

3.订阅器类

package com.shoshana.publishsubscribe;

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

public class SubscribePublish {

//订阅器名称

private String name;

//订阅器队列容量

final int QUEUE_CAPACITY = 20;

//订阅器存储队列

private BlockingQueue queue = new ArrayBlockingQueue(QUEUE_CAPACITY);

//订阅者

private List subcribers = new ArrayList();

public SubscribePublish(String name) {

this.name = name;

}

public void publish(String publisher, M message, boolean isInstantMsg) {

if (isInstantMsg) {

update(publisher, message);

return;

}

Msg m = new Msg(publisher, message);

if (!queue.offer(m)) {

update();

}

}

public void subcribe(ISubcriber subcriber) {

subcribers.add(subcriber);

}

public void unSubcribe(ISubcriber subcriber) {

subcribers.remove(subcriber);

}

public void update() {

Msg m = null;

while ((m = queue.peek()) != null) {

this.update(m.getPublisher(), (M) m.getMsg());

}

}

public void update(String publisher, M Msg) {

for (ISubcriber subcriber : subcribers) {

subcriber.update(publisher, Msg);

}

}

}

class Msg {

private String publisher;

private M m;

public Msg(String publisher, M m) {

this.publisher = publisher;

this.m = m;

}

public String getPublisher() {

return publisher;

}

public void setPublisher(String publisher) {

this.publisher = publisher;

}

public M getMsg() {

return m;

}

public void setMsg(M m) {

this.m = m;

}

}

4.发布者实现类

package com.shoshana.publishsubscribe;

public class PublisherImpOne implements IPublisher {

private String name;

public PublisherImpOne(String name) {

super();

this.name = name;

}

public void publish(SubscribePublish subscribePublish, M message, boolean isInstantMsg) {

subscribePublish.publish(this.name, message, isInstantMsg);

}

}

5.订阅者实现类

package com.shoshana.publishsubscribe;

public class SubcriberImpOne implements ISubcriber {

public String name;

public SubcriberImpOne(String name) {

super();

this.name = name;

}

public void subcribe(SubscribePublish subscribePublish) {

subscribePublish.subcribe(this);

}

public void unSubcribe(SubscribePublish subscribePublish) {

subscribePublish.unSubcribe(this);

}

public void update(String publisher, M message) {

System.out.println(this.name + "收到" + publisher + "发来的消息:" + message.toString());

}

}

6.测试类

package com.shoshana.publishsubscribe;

public class SubPubTest {

public static void main(String[] args) {

SubscribePublish subscribePublish = new SubscribePublish("订阅器");

IPublisher publisher1 = new PublisherImpOne("发布者1");

ISubcriber subcriber1 = new SubcriberImpOne("订阅者1");

ISubcriber subcriber2 = new SubcriberImpOne("订阅者2");

subcriber1.subcribe(subscribePublish);

subcriber2.subcribe(subscribePublish);

publisher1.publish(subscribePublish, "welcome", true);

publisher1.publish(subscribePublish, "to", true);

publisher1.publish(subscribePublish, "yy", false);

}

}

测试结果:

com.shoshana.publishsubscribe.SubPubTest

订阅者1收到发布者1发来的消息:welcome

订阅者2收到发布者1发来的消息:welcome

订阅者1收到发布者1发来的消息:to

订阅者2收到发布者1发来的消息:to

Process finished with exit code 0