Java Streams类代码示例

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› Java›Java编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Java中reactor.rx.Streams类的典型用法代码示例。如果您正苦于以下问题：Java Streams类的具体用法？Java Streams怎么用？Java Streams使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

Streams类属于reactor.rx包，在下文中一共展示了Streams类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。

示例1: equals

import reactor.rx.Streams; //导入依赖的package包/类
@Override
public boolean equals(Object o) {
    if (o == this)
        return true;
    if (!(o instanceof RedissonListReactive))
        return false;

    Stream<Object> e1 = Streams.wrap((Publisher<Object>)iterator());
    Stream<Object> e2 = Streams.wrap(((RedissonListReactive<Object>) o).iterator());
    Long count = Streams.merge(e1, e2).groupBy(new Function<Object, Object>() {
        @Override
        public Object apply(Object t) {
            return t;
        }
    }).count().next().poll();

    boolean res = count.intValue() == Streams.wrap(size()).next().poll();
    res &= count.intValue() == Streams.wrap(((RedissonListReactive<Object>) o).size()).next().poll();
    return res;
}

开发者ID:qq1588518，项目名称:JRediClients，代码行数:21，代码来源:RedissonListReactive.java

示例2: mapFunction

import reactor.rx.Streams; //导入依赖的package包/类
private Function<Entry<K, V>, Boolean> mapFunction(final RMapReactive<Object, Object> m) {
    return new Function<Map.Entry<K, V>, Boolean>() {
        @Override
        public Boolean apply(Entry<K, V> e) {
            Object key = e.getKey();
            Object value = e.getValue();
            if (value == null) {
                if (!(Streams.create(m.get(key)).next().poll() ==null && Streams.create(m.containsKey(key)).next().poll()))
                    return false;
            } else {
                if (!value.equals(Streams.create(m.get(key)).next().poll()))
                    return false;
            }
            return true;
        }
    };
}

开发者ID:qq1588518，项目名称:JRediClients，代码行数:18，代码来源:RedissonMapReactive.java

示例3: echoJsonStreamDecoding

import reactor.rx.Streams; //导入依赖的package包/类
private static void echoJsonStreamDecoding() {

        TcpServer<Person, Person> transport = Netty4TcpServer.<Person, Person>create(
                0,
                new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        channel.pipeline().addFirst(
                                new JsonObjectDecoder(),
                                new JacksonJsonCodec());
                    }
                });

        ReactorTcpServer.create(transport)
                .start(connection -> {
                    connection.log("input")
                            .observeComplete(v -> LOG.info("Connection input complete"))
                            .capacity(1)
                            .consume(person -> {
                                person = new Person(person.getLastName(), person.getFirstName());
                                Streams.wrap(connection.writeWith(Streams.just(person))).consume();
                            });
                    return Streams.never();
                });

    }

开发者ID:reactive-ipc，项目名称:reactive-ipc-jvm，代码行数:27，代码来源:CodecSample.java

示例4: getHandler

import reactor.rx.Streams; //导入依赖的package包/类
private ReactorChannelHandler<String, String, HttpChannel<String, String>> getHandler() {
	return channel -> {
		channel.headers()
				.entries()
				.forEach(
						entry1 -> System.out.println(String.format(
								"header [%s=>%s]", entry1.getKey(),
								entry1.getValue())));
		String response;
		try {
			response = getWebPage("src/main/java/webapp/ws.html");
		} catch (IOException e) {
			e.printStackTrace();
			response = e.getMessage();
		}
		System.out.println(String.format("%s from thread %s",
				response.toString(), Thread.currentThread()));
		return channel.writeWith(Streams.just(response
				.toString()));
	};
}

开发者ID:iproduct，项目名称:low-latency-high-throughput，代码行数:22，代码来源:ReactorWishesWS.java

示例5: postHandler

import reactor.rx.Streams; //导入依赖的package包/类
private ReactorChannelHandler<Buffer, Buffer, HttpChannel<Buffer, Buffer>> postHandler() {
	return channel -> {

		channel.headers().entries().forEach(entry -> System.out.println(String.format("header [%s=>%s]", entry.getKey(),
				entry.getValue())));

		return channel.writeWith(Streams
				.wrap(channel)
				.take(1)
				.log("received")
				.flatMap(data -> {
					final StringBuilder response = new StringBuilder().append("hello ").append(new String(data.asBytes()));
					System.out.println(String.format("%s from thread %s", response.toString(), Thread.currentThread()));
					return Streams.just(Buffer.wrap(response.toString()));
				}));
	};
}

开发者ID:iproduct，项目名称:low-latency-high-throughput，代码行数:18，代码来源:ReactorNetGetPost.java

示例6: main

import reactor.rx.Streams; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException {
	Environment env = Environment.initialize();
	CountDownLatch latch = new CountDownLatch(5);

	Dispatcher trreadPoolDispatcher = new ThreadPoolExecutorDispatcher(5, 128);

	Consumer<String> consumer = ev -> {
	    LOG.info("Hello " + ev + " from thread: " +Thread.currentThread() + "\n");
	    latch.countDown(); 
	};

	Consumer<Throwable> errorConsumer = error ->
	    error.printStackTrace();

	// a task is submitted to the thread pool dispatcher
	Stream<String> stream = Streams.just("One", "Two", "Three", "Four", "Five", "Six", "Seven");
	stream.dispatchOn(env).consume(ev -> {
				System.out.println(ev);
				trreadPoolDispatcher.dispatch(ev, consumer, errorConsumer);
			});

	latch.await(15, TimeUnit.SECONDS); // Wait for task to execute

}

开发者ID:iproduct，项目名称:low-latency-high-throughput，代码行数:25，代码来源:ThreaadPoolExecutorDispatcher.java

示例7: main

import reactor.rx.Streams; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException {
	Environment env = Environment.initialize();
	CountDownLatch latch = new CountDownLatch(5);

	DispatcherSupplier supplier = Environment.newCachedDispatchers(3, "myPool");
	
	Consumer<String> consumer = ev -> {
	    LOG.info("Hello " + ev + " from thread: " +Thread.currentThread() + "\n");
	    latch.countDown(); 
	};

	Consumer<Throwable> errorConsumer = error ->
	    error.printStackTrace();

	// a task is submitted to the thread pool dispatcher
	Stream<String> stream = Streams.just("One", "Two", "Three", "Four", "Five", "Six", "Seven");
	stream.dispatchOn(env).partition().consume(
		groupStream -> 
			groupStream.dispatchOn(supplier.get()).consume(consumer, errorConsumer)
	);
	latch.await(15, TimeUnit.SECONDS); // Wait for task to execute

}

开发者ID:iproduct，项目名称:low-latency-high-throughput，代码行数:24，代码来源:DispatcherSupplierPartitioning.java

示例8: main

import reactor.rx.Streams; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException {
	Environment env = Environment.initialize();

	Processor<String, String> p = RingBufferProcessor.create("testProcessor", 32); 
	Stream<String> s1 = Streams.wrap(p); 

	s1.consume(ev -> System.out.println(Thread.currentThread() + " data=" + ev)); 
	s1.consume(ev -> System.out.println(Thread.currentThread() + " data=" + ev)); 
	s1.consume(ev -> System.out.println(Thread.currentThread() + " data=" + ev)); 
	
	p.onNext("One");
	p.onNext("Two");
	p.onNext("Three");
	p.onComplete();
	
	Environment.terminate();

}

开发者ID:iproduct，项目名称:low-latency-high-throughput，代码行数:19，代码来源:RingBufferProcessorDemo.java

示例9: main

import reactor.rx.Streams; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException {
	Environment env = Environment.initialize();

	Processor<String, String> p = RingBufferWorkProcessor.create("testProcessor", 32); 
	Stream<String> s1 = Streams.wrap(p); 

	s1.consume(ev -> System.out.println(Thread.currentThread() + " data=" + ev)); 
	s1.consume(ev -> System.out.println(Thread.currentThread() + " data=" + ev)); 
	s1.consume(ev -> System.out.println(Thread.currentThread() + " data=" + ev)); 
	
	p.onNext("One");
	p.onNext("Two");
	p.onNext("Three");
	p.onNext("Four");
	p.onNext("Five");
	p.onComplete();
	
	Environment.terminate();

}

开发者ID:iproduct，项目名称:low-latency-high-throughput，代码行数:21，代码来源:RingBufferWorkProcessorDemo.java

示例10: getHandler

import reactor.rx.Streams; //导入依赖的package包/类
private ReactorChannelHandler<String, String, HttpChannel<String, String>> getHandler() {
		return channel -> {
//			channel.headers()
//					.entries()
//					.forEach(
//							entry1 -> System.out.println(String.format(
//									"header [%s=>%s]", entry1.getKey(),
//									entry1.getValue())));
//			System.out.println(channel.uri());
			String uri = channel.uri();
			if (uri.equals("/")) 
				uri = "/index.html";
			String path = "src/main/webapp" + uri;	
			
			String response;
			try {
				response = getStaticResource(path);
			} catch (IOException e) {
				e.printStackTrace();
				response = e.getMessage();
			}
		
			return channel.writeWith(Streams.just(response
					.toString()));
		};
	}

开发者ID:iproduct，项目名称:low-latency-high-throughput，代码行数:27，代码来源:ReactorWishesWS.java

示例11: process

import reactor.rx.Streams; //导入依赖的package包/类
@Override
public Stream<Tuple> process(Stream<String> stream) {


	return stream.flatMap(tweet -> {
		JSONArray array = JsonPath.read(tweet, "$.entities.hashtags[*].text");
		return Streams.from(array.toArray(new String[array.size()]));
	})
			.map(w -> reactor.fn.tuple.Tuple.of(w, 1))
			.window(timeWindow, SECONDS)
			.flatMap(s -> BiStreams.reduceByKey(s, (acc, next) -> acc + next)
					.sort((a, b) -> -a.t2.compareTo(b.t2))
					.take(topN))
			.map(entry -> tuple().of("hashtag", entry.t1, "count", entry.t2));

}

开发者ID:spring-projects，项目名称:spring-xd-samples，代码行数:17，代码来源:TopTags.java

示例12: getKeysByPattern

import reactor.rx.Streams; //导入依赖的package包/类
@Override
public Publisher<String> getKeysByPattern(final String pattern) {
    List<Publisher<String>> publishers = new ArrayList<Publisher<String>>();
    for (MasterSlaveEntry entry : commandExecutor.getConnectionManager().getEntrySet()) {
        publishers.add(createKeysIterator(entry, pattern));
    }
    return Streams.merge(publishers);
}

开发者ID:qq1588518，项目名称:JRediClients，代码行数:9，代码来源:RedissonKeysReactive.java

示例13: sync

import reactor.rx.Streams; //导入依赖的package包/类
public static <V> V sync(Publisher<V> ob) {
    Promise<V> promise;
    if (Promise.class.isAssignableFrom(ob.getClass())) {
        promise = (Promise<V>) ob;
    } else {
        promise = Streams.wrap(ob).next();
    }

    V val = promise.poll();
    if (promise.isError()) {
        throw new RuntimeException(promise.reason());
    }
    return val;
}

开发者ID:qq1588518，项目名称:JRediClients，代码行数:15，代码来源:BaseReactiveTest.java

示例14: createProcessor

import reactor.rx.Streams; //导入依赖的package包/类
@PostConstruct
private void createProcessor() {
    processor = RingBufferProcessor.create();
    Stream stream = Streams.wrap(processor);
    stream.buffer(1, TimeUnit.SECONDS).consume(new Consumer<List<Long>>() {
        @Override
        public void accept(List<Long> repositoryIds) {
            for (Long repositoryId : Sets.newHashSet(repositoryIds)) {
                setRepositoryStatsOutOfDate(repositoryId);
            }
        }
    });
}

开发者ID:box，项目名称:mojito，代码行数:14，代码来源:RepositoryStatisticsUpdatedReactor.java

示例15: writeMultipleValues

import reactor.rx.Streams; //导入依赖的package包/类
@Test
public void writeMultipleValues() throws IOException {
    Promise<ByteBuf> chunk1 = Promises.success(Unpooled.buffer().writeBytes("This is".getBytes()));
    Promise<ByteBuf> chunk2 = Promises.success(Unpooled.buffer().writeBytes(" a test!".getBytes()));
    reactorServer.start(connection -> connection.writeWith(Streams.concat(chunk1, chunk2)));
    assertEquals("This is a test!", SocketTestUtils.read("localhost", reactorServer.getPort()));
}

开发者ID:reactive-ipc，项目名称:reactive-ipc-jvm，代码行数:8，代码来源:ReactorTcpServerTests.java

示例16: echo

import reactor.rx.Streams; //导入依赖的package包/类
/**
 * Keep echoing until the client goes away.
 */
private static void echo(TcpServer<ByteBuf, ByteBuf> transport) {
    ReactorTcpServer.create(transport)
        .startAndAwait(connection -> {
            connection.flatMap(inByteBuf -> {
                String text = "Hello " + inByteBuf.toString(Charset.defaultCharset());
                ByteBuf outByteBuf = Unpooled.buffer().writeBytes(text.getBytes());
                return connection.writeWith(Streams.just(outByteBuf));
            }).consume();
            return Streams.never();
        });
}

开发者ID:reactive-ipc，项目名称:reactive-ipc-jvm，代码行数:15，代码来源:ReactorTcpServerSample.java

示例17: echoWithQuitCommand

import reactor.rx.Streams; //导入依赖的package包/类
/**
 * Keep echoing until the client sends "quit".
 */
private static void echoWithQuitCommand(TcpServer<ByteBuf, ByteBuf> transport) {
    ReactorTcpServer.create(transport)
        .start(connection -> connection
                .map(byteBuf -> byteBuf.toString(Charset.defaultCharset()))
                .takeWhile(input -> !"quit".equalsIgnoreCase(input.trim()))
                .filter(input -> !"quit".equalsIgnoreCase(input.trim()))
                .map(input -> "Hello " + input)
                .flatMap(text -> connection.writeWith(
                        Streams.just(Unpooled.buffer().writeBytes(text.getBytes()))
                    )
                )
                .after()
        );
}

开发者ID:reactive-ipc，项目名称:reactive-ipc-jvm，代码行数:18，代码来源:ReactorTcpServerSample.java

示例18: runLineBasedFrameDecoder

import reactor.rx.Streams; //导入依赖的package包/类
private static void runLineBasedFrameDecoder() {

        TcpServer<String, String> transport = Netty4TcpServer.<String, String>create(
                0,
                new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        int bufferSize = 1;
                        ChannelConfig config = channel.config();
                        config.setOption(ChannelOption.SO_RCVBUF, bufferSize);
                        config.setOption(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(bufferSize));
                        channel.pipeline().addFirst(
                                new LineBasedFrameDecoder(256),
                                new StringDecoder(CharsetUtil.UTF_8),
                                new StringEncoder(CharsetUtil.UTF_8));
                    }
                });

        ReactorTcpServer.create(transport).start(connection -> {
            connection.log("input")
                    .observeComplete(v -> LOG.info("Connection input complete"))
                    .capacity(1)
                    .consume(line -> {
                        String response = "Hello " + line + "\n";
                        Streams.wrap(connection.writeWith(Streams.just(response))).consume();
                    });
            return Streams.never();
        });
    }

开发者ID:reactive-ipc，项目名称:reactive-ipc-jvm，代码行数:30，代码来源:CodecSample.java

示例19: TaxiStream

import reactor.rx.Streams; //导入依赖的package包/类
public TaxiStream(String fileName) {
    this.fileName = fileName;

    // create a Processor with an internal RingBuffer capacity of 32 slots
    this.tripsProcessor = RingBufferProcessor.create("trips", 32);

    // create a Reactor Stream from this Reactive Streams Processor
    this.trips = Streams.wrap(this.tripsProcessor);
}

开发者ID:jeryini，项目名称:TaxiManagementSystem，代码行数:10，代码来源:TaxiStream.java

示例20: main

import reactor.rx.Streams; //导入依赖的package包/类
public static void main(String... args) throws InterruptedException {
  Environment.initialize(); 

  Streams.zip(
	Streams.range(0, 3),
	Streams.from(new String[]{"Hello", "from", "Reactor", "Websocket"}),
	t2 -> t2).throttle(3000)
.consume( t2 -> System.out.println(t2.getT1() + ": " + t2.getT2()));
	
  Thread.sleep(500); 
}

开发者ID:iproduct，项目名称:low-latency-high-throughput，代码行数:12，代码来源:ZipStreamsDemo.java

注：本文中的reactor.rx.Streams类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Java HttpUrlConnectorProvider类代码示例发布时间：2022-05-22

Java IndentingXMLStreamWriter类代码示例发布时间：2022-05-22

剪的笔顺,诠释剪的笔画,认识剪的部首

1 六六分期app的软件客服如何联系？(六六分期

六六分期app的软件客服如何联系？不知道吗？加qq群【895510560】即可！标题：六六分期

阅读：18033|2023-10-27

2 可心卡盟:win10系统火狐flash插件崩溃怎么

今天小编告诉大家如何处理win10系统火狐flash插件总是崩溃的问题，可能很多用户都不知

阅读：9595|2022-11-06

3 亲亲特价:怎么删除回收站图标

今天小编告诉大家如何对win10系统删除桌面回收站图标进行设置，可能很多用户都不知道

阅读：8141|2022-11-06

4 济南大学虚拟社区:鲁大师节能降温的具体办

今天小编告诉大家如何对win10系统电脑设置节能降温的设置方法，想必大家都遇到过需要

阅读：8522|2022-11-06

5 xlueops.exe:无线网络安装向导

我们在使用xp系统的过程中,经常需要对xp系统无线网络安装向导设置进行设置，可能很多

阅读：8425|2022-11-06

6 女斗合众国:win7系统cf与主机连接不稳定怎

今天小编告诉大家如何处理win7系统玩cf老是与主机连接不稳定的问题，可能很多用户都不

阅读：9333|2022-11-06

7 0xc000022-[cf烟雾头]cf怎么调烟雾头

电脑对日常生活的重要性小编就不多说了，可是一旦碰到win7系统设置cf烟雾头的问题，很

阅读：8390|2022-11-06

8 qizideyouhuo:应用程序无法正常启动0xc0000

我们在日常使用电脑的时候，有的小伙伴们可能在打开应用的时候会遇见提示应用程序无法

阅读：7826|2022-11-06

9 ipz-185:win7系统vcf文件怎么打开

今天小编告诉大家如何对win7系统打开vcf文件进行设置，可能很多用户都不知道怎么对win

阅读：8379|2022-11-06

10 傻哥蹦迪:win10系统s4怎么打开usb调试

今天小编告诉大家如何对win10系统s4开启USB调试模式进行设置，可能很多用户都不知道怎

阅读：7373|2022-11-06

客服电话

电子邮件

Java Streams类代码示例

示例1: equals

示例2: mapFunction

示例3: echoJsonStreamDecoding

示例4: getHandler

示例5: postHandler

示例6: main

示例7: main

示例8: main

示例9: main

示例10: getHandler

示例11: process

示例12: getKeysByPattern

示例13: sync

示例14: createProcessor

示例15: writeMultipleValues

示例16: echo

示例17: echoWithQuitCommand

示例18: runLineBasedFrameDecoder

示例19: TaxiStream

示例20: main

请发表评论

全部评论

上一篇：

下一篇：

CVE-2022-1626

librespeed/speedtest: Self-hosted Speedt

freeCodeCamp/boilerplate-project-library

avehtari/BDA_m_demos: Bayesian Data Anal

四维彩超怎么看性别？四维看男孩女孩诀窍

剪的笔顺,诠释剪的笔画,认识剪的部首

六六分期app的软件客服如何联系？(六六分期

florent37/ViewAnimator: A fluent Android

florent37/Shrine-MaterialDesign2: implem

CVE-2020-36276

SimpleSoftwareIO/simple-sms: Send and re

关于我们

产品与服务

解决方案

139-2527-9053