Commit 0a2fb6e

committed

在此多处细节迭代

1 parent 99059dc commit 0a2fb6eCopy full SHA for 0a2fb6e

File tree

+14

-9

lines changed

chapter13 Reactive Streams/java9-reactive-demo1/src/main/java/com/dockerx/reactive/jdkaction

+14

-9

lines changed

Lines changed: 2 additions & 2 deletions

Original file line number	Diff line number	Diff line change
`@@ -14,7 +14,7 @@`
`14`	`14`	`public class DockerXDemoApplication {`
`15`	`15`
`16`	`16`	`public static void main(String[] args) {`
`17`		`- Flow_submissionPublisher();`
	`17`	`+ Flow_customsubmissionPublisher();`
`18`	`18`	`}`
`19`	`19`
`20`	`20`	`private static void demoSubscribe(DockerXDemoPublisher<Integer> publisher, String subscriberName){`
`@@ -23,7 +23,7 @@ private static void demoSubscribe(DockerXDemoPublisher<Integer> publisher, Strin`
`23`	`23`	`publisher.subscribe(subscriber);`
`24`	`24`	`}`
`25`	`25`
`26`		`- private static void Flow_submissionPublisher() {`
	`26`	`+ private static void Flow_customsubmissionPublisher() {`
`27`	`27`	`ExecutorService execService = ForkJoinPool.commonPool();//Executors.newFixedThreadPool(3);`
`28`	`28`	`try (DockerXDemoPublisher<Integer> publisher = new DockerXDemoPublisher<>()){`
`29`	`29`	`demoSubscribe(publisher, "One");`

Lines changed: 4 additions & 2 deletions

Original file line number	Diff line number	Diff line change
`@@ -13,14 +13,16 @@ public class DockerXDemoPublisher<T> implements Flow.Publisher<T>, AutoCloseable`
`13`	`13`
`14`	`14`
`15`	`15`	`public void submit(T item) {`
	`16`	`+ System.out.println("*************** 开始发布元素 item: "+item+" ***************");`
`16`	`17`	`list.forEach(e -> {`
`17`		`- e.subscriber.onNext(item);`
	`18`	`+ e.future=executor.submit(() -> { e.subscriber.onNext(item);});`
	`19`	`+`
`18`	`20`	`});`
`19`	`21`	`}`
`20`	`22`
`21`	`23`	`public void close() {`
`22`	`24`	`list.forEach(e -> {`
`23`		`- e.subscriber.onComplete();`
	`25`	`+ e.future=executor.submit(() -> { e.subscriber.onComplete();});`
`24`	`26`	`});`
`25`	`27`	`}`
`26`	`28`

Lines changed: 8 additions & 5 deletions

Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ public DockerXDemoSubscriber(long bufferSize, String name) {`
`27`	`27`	`}`
`28`	`28`
`29`	`29`	`public void onSubscribe(Flow.Subscription subscription) {`
`30`		`- count = bufferSize - bufferSize / 2;// 当消费一半的时候重新请求`
	`30`	`+ //count = bufferSize - bufferSize / 2;// 当消费一半的时候重新请求`
`31`	`31`	`(this.subscription = subscription).request(bufferSize);`
`32`	`32`	`System.out.println("开始onSubscribe订阅");`
`33`	`33`	`try {`
`@@ -38,11 +38,14 @@ public void onSubscribe(Flow.Subscription subscription) {`
`38`	`38`	`}`
`39`	`39`
`40`	`40`	`public void onNext(T item) {`
`41`		`- if (--count <= 0) subscription.request(count = bufferSize - bufferSize / 2);`
`42`		`- System.out.println(" ############# " + Thread.currentThread().getName()+" "+name+" "+item+" "+count+ " ############# ");`
	`41`	`+ //if (--count <= 0) subscription.request(count = bufferSize - bufferSize / 2);`
	`42`	`+ System.out.println(" ############# " + Thread.currentThread().getName()+" name: "+name+" item: "+item+ " ############# ");`
`43`	`43`	`System.out.println(name + " received: " + item);`
`44`		`-`
`45`		`-`
	`44`	`+ try {`
	`45`	`+ Thread.sleep(10);`
	`46`	`+ } catch (InterruptedException e) {`
	`47`	`+ e.printStackTrace();`
	`48`	`+ }`
`46`	`49`
`47`	`50`	`}`
`48`	`51`

Comments

(0)