パブリッシャーとサブスクライブ

この辺の理解が少し曖昧なのですが、パブリッシャーはFluxやMonoのことです。

public abstract class Flux<T> implements CorePublisher<T> {

これらは対象とするソースからノンブロッキング処理を行うことができます

しかしパブリッシャーはサブスクライブしないと何もしませんそれを確認するためにテストをしてみます

以下は前回作ったコードです

public class UserDao {
	
	public Flux<User> getUsers(){
		return Flux.range(1, 10)
				.delayElements(Duration.ofSeconds(1))
				.doFirst(this::first)
				.doOnNext(this::writeLog)
				.doOnComplete(this::complete)
				.map(this::createUser);
	}
	
	private void first() {
		System.out.println("処理を開始します");
	}
	
	private void complete() {
		System.out.println("処理が終了しました");
	}
	
	private User createUser(int i) {
		return new User("sampleUser", i);
	}
	
	private void writeLog(int i) {
		System.out.println("done " + String.valueOf(i) + " .");
	}
}

getUsersメソッドでFluxを取得しています
処理が実行されると様々なログが出力されます

このメソッドを利用して取得してみます

class UserDaoTest {
	
	private UserDao userDao;
	
	@BeforeEach
	private void prepare() {
		this.userDao = new UserDao();
	}
	
	@Test
	public void test() {
			Flux<User> users = this.userDao.getUsers();
	}

}

しかしログには何も表示されていません。

なぜかというサブスクライブしていないからです。今度は明示的にsubscribeメソッドを呼び出します。

@Test
public void test() {
		Flux<User> users = this.userDao.getUsers();
		users.subscribe();
}

テストが終わってしまうのでログが最初のログしか流れていませんが、subscribeすることでパブリッシャーが実行することがわかりました。

デバッグして処理を止めておくとログが次々に流れていくのが確認できます。

確認不足ですが、Springでは以下の個所で呼ばれているのか？

return ServerResponse.ok().
	contentType(MediaType.TEXT_EVENT_STREAM).
	body(users,User.class);

とりあえずよくあるサンプル的なものでは明示的にsubscribeを呼び出してはいませんでした。

ノンブロッキング

スレッドがブロッキングされることなく処理が進んでいくかどうかを見ていきます

パブリッシャーのdoFirstでスレッド名を表示させるようにします

private void first() {
	System.out.println(Thread.currentThread().getName());
	System.out.println("処理を開始します");
}

先ほどのdaoクラスから複数のFluxを作成して、同時にsubscribeメソッドをコールしてみます

@Component
public class TestHandle {
	
	@Autowired
	private UserDao dao;
	
	public Mono<ServerResponse> getUsers(ServerRequest request){
		
		System.out.println(Thread.currentThread().getName() + "　UserDao#getUsersを実行します");
		
		Flux<User> users = this.dao.getUsers();
		Flux<User> users2 = this.dao.getUsers();
		Flux<User> users3 = this.dao.getUsers();
	
		users2.subscribe();
		users3.subscribe();
		
		return ServerResponse.ok().
				contentType(MediaType.TEXT_EVENT_STREAM).
				body(users,User.class);
	}

}

画面から確認してみると以下のように、同じスレッド内でブロッキングされずにログが出力されていることが確認できます

reactor-http-nio-2　UserDao#getUsersを実行します
reactor-http-nio-2
処理を開始します
reactor-http-nio-2
処理を開始します
reactor-http-nio-2
処理を開始します
done 1 .
done 1 .
done 1 .
done 2 .
done 2 .
done 2 .
done 3 .
done 3 .
done 3 .
done 4 .
done 4 .
done 4 .
done 5 .
done 5 .
done 5 .
done 6 .
done 6 .
done 6 .
done 7 .
done 7 .
done 7 .
done 8 .
done 8 .
done 8 .
done 9 .
done 9 .
done 9 .
done 10 .
done 10 .
処理が終了しました
処理が終了しました
done 10 .
処理が終了しました

同時に複数の処理が進んでいることが確認できました。今回はこの辺にしておきます。

Spring WebFlux | パブリッシャーとサブスクライブ