多线程面试题 消费者（“怎么洞察消费者需求”这个面试题请问该如何回答）

本文目录

• “怎么洞察消费者需求”这个面试题请问该如何回答
• java实现生产者和消费者问题的几种方式
• 面试官杠上重复消费、消息堆积、消息丢失、顺序消息
• 操作系统简答题
• 由生产者/消费者问题看JAVA多线程

“怎么洞察消费者需求”这个面试题请问该如何回答

既然通过消费者洞察抓住消费者的真实需求有着强大的市场效应,那么什么才是好的消费者洞察呢?...
2.消费者洞察是一种动态的研究,也就是动态跟踪消费者为什么会购买我们的产品、为什么会...
3.值得注意的是,有深度的消费者洞察不能只是研究消费者表面的行为,还要有看穿表面的能力,...
4.要做好品牌营销中消费者洞察,在很大程度上需要不断的实践积累和实践经验,...
5.可以说,世上没有一个品牌能说它已经完全了解消费者,并且很好地掌握并占领了市场,...

java实现生产者和消费者问题的几种方式

生产者消费者问题是多线程的一个经典问题，它描述是有一块缓冲区作为仓库，生产者可以将产品放入仓库，消费者则可以从仓库中取走产品。
解决生产者/消费者问题的方法可分为两类：
采用某种机制保护生产者和消费者之间的同步；
在生产者和消费者之间建立一个管道。
第一种方式有较高的效率，并且易于实现，代码的可控制性较好，属于常用的模式。第二种管道缓冲区不易控制，被传输数据对象不易于封装等，实用性不强。
在Java中有四种方法支持同步，其中前三个是同步方法，一个是管道方法。
wait()
/
notify()方法
await()
/
signal()方法
BlockingQueue阻塞队列方法
PipedInputStream
/
PipedOutputStream
通过
wait()
/
notify()方法实现：
wait()
/
nofity()方法是基类Object的两个方法：
wait()方法：当缓冲区已满/空时，生产者/消费者线程停止自己的执行，放弃锁，使自己处于等等状态，让其他线程执行。
notify()方法：当生产者/消费者向缓冲区放入/取出一个产品时，向其他等待的线程发出可执行的通知，同时放弃锁，使自己处于等待状态。
通过await()
/
signal()方法实现：
await()和signal()的功能基本上和wait()
/
nofity()相同，完全可以取代它们，但是它们和新引入的锁定机制Lock直接挂钩，具有更大的灵活性。通过在Lock对象上调用newCondition()方法，将条件变量和一个锁对象进行绑定，进而控制并发程序访问竞争资源的安全。
通过BlockingQueue方法实现：
它是一个已经在内部实现了同步的队列，实现方式采用的是我们第2种await()
/
signal()方法。它可以在生成对象时指定容量大小。它用于阻塞操作的是put()和take()方法：
put()方法：类似于我们上面的生产者线程，容量达到最大时，自动阻塞。
take()方法：类似于我们上面的消费者线程，容量为0时，自动阻塞。

面试官杠上重复消费、消息堆积、消息丢失、顺序消息

消息队列在互联网技术存储方面使用如此广泛，几乎所有的后端技术面试官都要在消息队列的使用和原理方面对小伙伴们进行360 的刁难。

什么，这么多问题啊！别慌，现在就来找找解决方案。

一、 重复消费

现在消息队列一般都能保证at least once的，也就是消息至少一次投递。 在这种情况为什么会出现重复消费的问题呢？通常都是由于网络原因造成的 ，原因如下：通常消息被成功消费后消费者都会发送一个成功标志给MQ，MQ收到这个标志就表示消息已经成功消费了，就不会再发送给其他消费者了。但是如果因为网络这个标志没有送到MQ就丢失了，MQ就认为这个消息没有被成功消费，就会再次发送给其他消费者消费，就造成重复了。

这时我们看这个问题就变成了我们怎么保证消费端的幂等性。

幂等性 是指一个操作其执行任意多次所产生的影响均与一次执行的影响相同，大白话就是你同样的参数调用我这个接口，调用多少次结果都相同。

怎么保证消息队列消费的幂等性

其实还是得结合业务来思考，我在这里给出几个解决方案：

1. 分布式锁 。生产者发送每条数据的时候，里面加一个全局唯一的 id，类似订单 id 之类的东西，然后你这里消费到了之后，先根据这个 id 去比如 Redis 里查一下，之前消费过吗？如果没有消费过，你就处理，然后就是这个 id 写 Redis。如果消费过了，那你就别处理了，保证别重复处理相同的消息即可。

2.唯一键防重 。基于数据库的唯一键来保证重复数据不会重复插入多条。因为有唯一键约束了，重复数据插入只会报错，不会导致数据库中出现脏数据。

3.先查后写。 要写数据库前，先根据主键查一下，如果这数据都有了，你就别插入了，update一下好了。

4. 关闭重试机制 。如果把重试机制关掉的话不显示，虽然解决了重复消费的问题，但是可能会造成丢失消息，不建议这么做。

不同的业务可以选择不同的方案，如果服务的并发量不高，可以考虑唯一键防重或者先查后写的方案；如果并发量较高，追求性能，沐子推荐采用分布式锁实现幂等性（本公司目前采用的方案）

二、 消息堆积

1. 消息堆积的产生原因

消息堆积的原因主要在于两方面，其一为消费的太慢或消费方出现异常，其二为生产方生产的太快，总的来说就是 消息的速度赶不上生产的速度，生产和消费速度不匹配造成的 。

2. 消息堆积的解决方案

1）生产端：一般当生产端发生积压（Broker正常的情况下）就要查看你的业务逻辑是否有异常的耗时步骤导致的，是否需要改并行化操作等。

Broker端：当Broker端发生积压我们首先要查看，消息队列内存使用情况， 如果有分区的的话还得看每个分区积压的消息数量差异。当每个分区的消息积压数据量相对均匀的话，我们大致可以认为是流量激增。需要在消费端做优化，或者同时需要增加Broker节点（相当于存储扩容），如果分区加压消息数量差异很大的话（有的队列满了，有的队列可能还是空闲状态），我们这时候就要检查我们的路由转发规则是否合理。

2） 增加消费者 ，多部署几台消费者机器（横向扩展），提升消费者的消费能力。

3）此种情况可以将这些 消费不成功的消息转发到其它队列里去(类似死信队列) ，后面再慢慢分析死信队列里的消息处理问题。

4) mq 中的消息过期失效了。可以采取一个方案，就是批量重导 ，这个我们之前线上也有类似的场景干过。就是大量积压的时候，我们当时就直接丢弃数据了，然后等过了高峰期以后，将丢失的那批数据，写个临时程序，一点一点的查出来，然后重新灌入 mq 里面去，把白天丢的数据给他补回来。

总之，上面说到消息积压的问题，我们需要查看是否有无限重发的消息或者有进入死锁的程序等等，当确定是流量激增的话，我们需要评估是否需要增加资源还是通过限流的方式解决，当短时间大量消息需要处理时，在资源允许的情况下，我们可以新启一批消费者与消息队列，将原来的消费者中的消息直接作为生产者转发到临时应急队列中，这样大概率的能够快速解决消息积压。与其事后处理不如我们在设计之初就要把积压考虑进来，对于数据量非常大，但是实时性要求不高的场景，可以设计出批量消息发送，当队列积累到一定阀值再做批量消费消费，这里需要注意的就是重复消费带来的影响，设计不好就是一场灾难。

三、 消息丢失

一般来讲消息丢失的途径有三个： 生产者弄丢数据、消息队列弄丢数据、消费者弄丢数据 。

1. 生产者弄丢数据

a、丢失的原因：因为网络传输的不稳定性，当生产者在向MQ发送消息的过程中，MQ没有成功接收到消息，但是生产者却以为MQ成功接收到了消息，不会再次重复发送该消息，从而导致消息的丢失。

b、解决办法：有两个解决办法，第一个方法： 向broker发送消息时，如果由于网络抖动等原因导致消息发送失败，可以设置 失败重试次数让消息重发 。
第二个方法： 事务机制和confirm机制，最常用的是confirm机制 ；

事务机制和 confirm 机制最大的不同在于，事务机制是同步的，你提交一个事务之后会阻塞在那儿，但是 confirm 机制是异步的，你发送个消息之后就可以发送下一个消息，然后那个消息 MQ 接收了之后会异步回调你的一个接口通知你这个消息接收到了。

2. MQ弄丢数据

a、丢失的原因：MQ接收到生产者发送过来的消息，是存在内存中的，如果没有被消费完，此时MQ宕机了，那么再次启动的时候，原来内存中的那些消息都丢失了。

b、解决办法： 开启MQ的持久化 。结合上面的说到的confirm机制，只有当消息成功持久化磁盘之后，才会回调生产者的接口返回ack消息，否则都算失败，生产者会重新发送。存入磁盘的消息不会丢失，就算MQ挂掉了，重启之后，他会读取磁盘中的消息，不会导致消息的丢失。

注意，哪怕是你给 MQ 开启了持久化机制，也有一种可能，就是这个消息写到了MQ 中，但是还没来得及持久化到磁盘上，结果不巧，此时MQ挂了，就会导致内存里的一点点数据丢失。

所以，持久化可以跟生产者那边的 confirm 机制配合起来，只有消息被持久化到磁盘之后，才会通知生产者 ack 了，所以哪怕是在持久化到磁盘之前，MQ 挂了，数据丢了，生产者收不到 ack，你也是可以自己重发的。

3. 消费者弄丢数据

a、丢失的原因：如果MQ成功的把消息发送给了消费者，那么MQ的ack机制会自动的返回成功，表明发送消息成功，下次就不会发送这个消息。但如果就在此时，消费者还没处理完该消息，然后宕机了，那么这个消息就丢失了。

b、解决的办法：1）简单来说，就是必须关闭 MQ 的自动提交，把 自动提交改为手动提交 ，也就是说当我消费成功后才会进行提交。

2）消费者端已经正常接收到消息但是在执行后续消息处理时发生了异常，最终返回处理失败。 重试-进行重新消费问题，如果一直这样重复消费都持续失败到一定次数，可以投递到 DLQ 死信队列，应用可以监控死信队列来做人工干预 。

四、 顺序消费

比如一个电商的下单操作，下单后先减库存然后生成订单，这个操作就需要顺序执行的。队列本身是有顺序的，但是为什么还要保证顺序消费呢，主要是因为生产环境服务实例一般都是集群，当消费者是多个实例时，队列中的消息会分发到所有实例进行消费（同一个消息只能发给一个消费者实例），这样就不能保证消息顺序的消费，因为你不能确保哪台机器执行消费端业务代码的速度快。

保证每次只有单个消费实例消费

所以对于需要保证顺序消费的业务，我们可以只部署一个消费者实例，然后设置MQ 每次只推送一个消息，再开启手动 ack 即可。这样MQ 每次只会从队列推送一个消息过来，处理完成之后我们 ack 回应，再消费下一个，就能确保消息顺序性。

这样MQ 每次只会从队列推送一个消息过来，处理完成之后我们 ack 回应，再消费下一个，就能确保消息顺序性。

但是这样的操作也会降低消费者的性能， 一个消费者消费消息时，其他消费者会阻塞，所以很多场景下可能并不会采用这样的方案。

所以一般会根据场景，制定一定的策略来解决消费顺序问题。

多线程并发抢占出现消费乱序问题

当MQ采用简单队列模式的时候,如果消费者采用多线程的方式来加速消息的处理,此时也会出现消息乱序的问题。

多线程并发抢占出现消费乱序问题，将消息ID进行hash计算，将相同值放入同一个内存队列，让指定线程执行，即可解决顺序消费问题。

在多个分区中保证消息顺序和消息处理效率

首先使用多个分区，消息可以被发送端发送至多个分区，保证消息发送的效率。然后在消费端在拉消息时使用ConutdownLunch来记录一组有序消息的个数。如果达到个数，说明已拉取到完整的一组有序消息。然后在消费端根据消息序号进行排序，消费端将排好序的消息发到内存队列(可以搞多个)，一个内存队列开启一个线程顺序处理消息。即可最大程度上既保证顺序又保证效率！

RocketMQ作为阿里开源的一款高性能、高吞吐量的消息中间件，支持顺序消息，所以如果有这种场景或者要使用MQ，我建议你直接使用RocketMQ即可。

我们说了一些处理与分析问题的方法，这里有一个最重要的点就是我们需要有一套实用的监控发现工具或者方式，在问题第一时间发现才是王道，不然我们上面所说的都空谈，当问题发现的时候损失已经无法挽回。所以我们要在设计系统之初需要要为监控系统或者程序提供完备或者必须的日志，接口，数据等，这要才是一个合理的设计。当没有监控系统的情况下我们必须自己设计一套简单分析接口。

最后， 如果我的文章对你有所帮助或者有所启发， 欢迎关注公众号（微信搜索公众号：首席架构师专栏），里面有许多技术干货，也有我对技术的思考和感悟，还有作为架构师的验验分享；关注后回复 【面试题】，有我准备的面试题、架构师大型项目实战视频等福利 ， 我会带着你一起学习、成长，让我们一起加油！！！

操作系统简答题

生产者和消费者问题是从操作系统中的许多实际同步问题中抽象出来的具有
代表性的问题。它反映了操作系统中典型的同步例子。
　　生产者进程(进程由多个线程组成)生产信息，例如它可以是计算进程。消费
者进程使用信息，它可以是输出打印进程。由于生产者和消费者彼此独立，且运
行速度不确定，所以很可能出现生产者已产生了信息而消费者却没有来得及接受
信息这种情况。为此，需要引入由一个或者若干个存储单元组成的临时存储区，
以便存放生产者所产生的信息，平滑进程间由于速度不确定所带来的问题。这个
临时存储区叫做缓冲区，通常用一维数组来表示。
　　由一个或若干个存储单元组成的缓冲区叫作“有穷缓冲区”。下面我们来分
析一下有穷缓冲的生产者和消费者的例子。
　　假设有多个生产者和多个消费者，它们共享一个具有n个存储单元的有穷缓冲
区Buffer(0……n-1)，这是一个环形队列。其队尾指针Rear指向当前信息应存放
的位置(Buffer)，队首指针Front指向当前取出信息的位置(Buffer[front
])。生产者进程总是把信息存放在Buffer中，消费者进程则总是从Buffer
中取出信息。如果想使生产者进程和消费者进程协调合作，则必须使它们
遵循如下规则:
　　1) 只要缓冲区有存储单元，生产者都可往其中存放信息；当缓冲区已满时，
若任意生产者提出写要求，则都必须等待；
　　2) 只要缓冲区中有消息可取，消费者都可从缓冲区中取出消息；当缓冲区为
空时，若任意消费者想取出信息，则必须等待；
　　3) 生产者们和消费者们不能同时读、写缓冲区。
生产者消费者问题是研究多线程程序时绕不开的问题，它的描述是有一块生产者和消费者共享的有界缓冲区，生产者往缓冲区放入产品，消费者从缓冲区取走产品，这个过程可以无休止的执行，不能因缓冲区满生产者放不进产品而终止，也不能因缓冲区空消费者无产品可取而终止。
解决生产者消费者问题的方法有两种，一种是采用某种机制保持生产者和消费者之间的同步，一种是在生产者和消费者之间建立一个管道。前一种有较高的效率并且可控制性较好，比较常用，后一种由于管道缓冲区不易控制及被传输数据对象不易封装等原因，比较少用。
同步问题的核心在于，CPU是按时间片轮询的方式执行程序，我们无法知道某一个线程是否被执行、是否被抢占、是否结束等，因此生产者完全可能当缓冲区已满的时候还在放入产品，消费者也完全可能当缓冲区为空时还在取出产品。
现在同步问题的解决方法一般是采用信号或者加锁机制，即生产者线程当缓冲区已满时放弃自己的执行权，进入等待状态，并通知消费者线程执行。消费者线程当缓冲区已空时放弃自己的执行权，进入等待状态，并通知生产者线程执行。这样一来就保持了线程的同步，并避免了线程间互相等待而进入死锁状态。

由生产者/消费者问题看JAVA多线程

　　生产者消费者问题是研究多线程程序时绕不开的问题 它的描述是有一块生产者和消费者共享的有界缓冲区 生产者往缓冲区放入产品 消费者从缓冲区取走产品 这个过程可以无休止的执行 不能因缓冲区满生产者放不进产品而终止 也不能因缓冲区空消费者无产品可取而终止

　　解决生产者消费者问题的方法有两种 一种是采用某种机制保持生产者和消费者之间的同步 一种是在生产者和消费者之间建立一个管道 前一种有较高的效率并且可控制性较好 比较常用 后一种由于管道缓冲区不易控制及被传输数据对象不易封装等原因 比较少用

　　同步问题的核心在于 CPU是按时间片轮询的方式执行程序 我们无法知道某一个线程是否被执行 是否被抢占 是否结束等 因此生产者完全可能当缓冲区已满的时候还在放入产品 消费者也完全可能当缓冲区为空时还在取出产品

　　现在同步问题的解决方法一般是采用信号或者加锁机制 即生产者线程当缓冲区已满时放弃自己的执行权 进入等待状态 并通知消费者线程执行 消费者线程当缓冲区已空时放弃自己的执行权 进入等待状态 并通知生产者线程执行 这样一来就保持了线程的同步 并避免了线程间互相等待而进入死锁状态

　　JAVA语言提供了独立于平台的线程机制 保持了 write once run anywhere 的特色 同时也提供了对同步机制的良好支持

　　在JAVA中 一共有四种方法支持同步 其中三个是同步方法 一个是管道方法

　　 方法wait()/notify()

　　 方法await()/signal()

　　 阻塞队列方法BlockingQueue

　　 管道方法PipedInputStream/PipedOutputStream

　　下面我们看各个方法的实现

　　 方法wait()/notify()

　　wait()和notify()是根类Object的两个方法 也就意味着所有的JAVA类都会具有这个两个方法 为什么会被这样设计呢？我们可以认为所有的对象默认都具有一个锁 虽然我们看不到 也没有办法直接操作 但它是存在的

　　wait()方法表示 当缓冲区已满或空时 生产者或消费者线程停止自己的执行 放弃锁 使自己处于等待状态 让另一个线程开始执行

　　notify()方法表示 当生产者或消费者对缓冲区放入或取出一个产品时 向另一个线程发出可执行通知 同时放弃锁 使自己处于等待状态

　　下面是一个例子代码

　　import java util LinkedList;

　　public class Sycn {

　　private LinkedList《Object》 myList =new LinkedList《Object》();

　　private int MAX = ;

　　public Sycn (){

　　}

　　public void start(){

　　new Producer() start();

　　new Consumer() start();

　　}

　　public static void main(String args) throws Exception{

　　Sycn s = new Sycn ();

　　s start();

　　}

　　class Producer extends Thread{

　　public void run(){

　　while(true){

　　synchronized(myList){

　　try{

　　while(myList size() == MAX){

　　System out println( warning: it s full! );

　　myList wait();

　　}

　　Object o = new Object();

　　if(myList add(o)){

　　System out println( Producer: + o);

　　myList notify();

　　}

　　}catch(InterruptedException ie){

　　System out println( producer is interrupted! );

　　}

　　}

　　}

　　}

　　}

　　class Consumer extends Thread{

　　public void run(){

　　while(true){

　　synchronized(myList){

　　try{

　　while(myList size() == ){

　　System out println( warning: it s empty! );

　　myList wait();

　　}

　　Object o = myList removeLast();

　　System out println( Consumer: + o);

　　myList notify();

　　}catch(InterruptedException ie){

　　System out println( consumer is interrupted! );

　　}

　　}

　　}

　　}

　　}

　　}

　　 方法await()/signal()

　　在JDK 以后 JAVA提供了新的更加健壮的线程处理机制 包括了同步 锁定 线程池等等 它们可以实现更小粒度上的控制 await()和signal()就是其中用来做同步的两种方法 它们的功能基本上和wait()/notify()相同 完全可以取代它们 但是它们和新引入的锁定机制Lock直接挂钩 具有更大的灵活性

　　下面是一个例子代码

　　import java util LinkedList;

　　import ncurrent locks *;

　　public class Sycn {

　　private LinkedList《Object》 myList = new LinkedList《Object》();

　　private int MAX = ;

　　private final Lock lock = new ReentrantLock();

　　private final Condition full = lock newCondition();

　　private final Condition empty = lock newCondition();

　　public Sycn (){

　　}

　　public void start(){

　　new Producer() start();

　　new Consumer() start();

　　}

　　public static void main(String args) throws Exception{

　　Sycn s = new Sycn ();

　　s start();

　　}

　　class Producer extends Thread{

　　public void run(){

　　while(true){

　　lock lock();

　　try{

　　while(myList size() == MAX){

　　System out println( warning: it s full! );

　　full await();

　　}

　　Object o = new Object();

　　if(myList add(o)){

　　System out println( Producer: + o);

　　empty signal();

　　}

　　}catch(InterruptedException ie){

　　System out println( producer is interrupted! );

　　}finally{

　　lock unlock();

　　}

　　}

　　}

　　}

　　class Consumer extends Thread{

　　public void run(){

　　while(true){

　　lock lock();

　　try{

　　while(myList size() == ){

　　System out println( warning: it s empty! );

　　empty await();

　　}

　　Object o = myList removeLast();

　　System out println( Consumer: + o);

　　full signal();

　　}catch(InterruptedException ie){

　　System out println( consumer is interrupted! );

　　}finally{

　　lock unlock();

　　}

　　}

　　}

　　}

　　}

　　 阻塞队列方法BlockingQueue

　　BlockingQueue也是JDK 的一部分 它是一个已经在内部实现了同步的队列 实现方式采用的是我们的第 种await()/signal()方法 它可以在生成对象时指定容量大小

　　它用于阻塞操作的是put()和take()方法

　　put()方法类似于我们上面的生产者线程 容量最大时 自动阻塞

　　take()方法类似于我们上面的消费者线程 容量为 时 自动阻塞

　　下面是一个例子代码

　　import ncurrent *;

　　public class Sycn {

　　private LinkedBlockingQueue《Object》 queue = new LinkedBlockingQueue《Object》( );

　　private int MAX = ;

　　public Sycn (){

　　}

　　public void start(){

　　new Producer() start();

　　new Consumer() start();

　　}

　　public static void main(String args) throws Exception{

　　Sycn s = new Sycn ();

　　s start();

　　}

　　class Producer extends Thread{

　　public void run(){

　　while(true){

　　//synchronized(this){

　　try{

　　if(queue size() == MAX)

　　System out println( warning: it s full! );

　　Object o = new Object();

　　queue put(o);

　　System out println( Producer: + o);

　　}catch(InterruptedException e){

　　System out println( producer is interrupted! );

　　}

　　//}

　　}

　　}

　　}

　　class Consumer extends Thread{

　　public void run(){

　　while(true){

　　//synchronized(this){

　　try{

　　if(queue size() == )

　　System out println( warning: it s empty! );

　　Object o = queue take();

　　System out println( Consumer: + o);

　　}catch(InterruptedException e){

　　System out println( producer is interrupted! );

　　}

　　//}

　　}

　　}

　　}

　　}

　　你发现这个例子中的问题了吗？

　　如果没有 我建议你运行一下这段代码 仔细观察它的输出 是不是有下面这个样子的？为什么会这样呢？

　　…

　　warning: it s full!

　　Producer: java lang object@ e a

　　…

　　你可能会说这是因为put()和System out println()之间没有同步造成的 我也这样认为 我也这样认为 但是你把run()中的synchronized前面的注释去掉 重新编译运行 有改观吗？没有 为什么？

　　这是因为 当缓冲区已满 生产者在put()操作时 put()内部调用了await()方法 放弃了线程的执行 然后消费者线程执行 调用take()方法 take()内部调用了signal()方法 通知生产者线程可以执行 致使在消费者的println()还没运行的情况下生产者的println()先被执行 所以有了上面的输出 run()中的synchronized其实并没有起什么作用

　　对于BlockingQueue大家可以放心使用 这可不是它的问题 只是在它和别的对象之间的同步有问题

　　对于这种多重嵌套同步的问题 以后再谈吧 欢迎大家讨论啊！

　　 管道方法PipedInputStream/PipedOutputStream

　　这个类位于java io包中 是解决同步问题的最简单的办法 一个线程将数据写入管道 另一个线程从管道读取数据 这样便构成了一种生产者/消费者的缓冲区编程模式

　　下面是一个例子代码 在这个代码我没有使用Object对象 而是简单的读写字节值 这是因为PipedInputStream/PipedOutputStream不允许传输对象 这是JAVA本身的一个bug 具体的大家可以看sun的解释 _bug do?bug_id=

　　import java io *;

　　public class Sycn {

　　private PipedOutputStream pos;

　　private PipedInputStream pis;

　　//private ObjectOutputStream oos;

　　//private ObjectInputStream ois;

　　public Sycn (){

　　try{

　　pos = new PipedOutputStream();

　　pis = new PipedInputStream(pos);

　　//oos = new ObjectOutputStream(pos);

　　//ois = new ObjectInputStream(pis);

　　}catch(IOException e){

　　System out println(e);

　　}

　　}

　　public void start(){

　　new Producer() start();

　　new Consumer() start();

　　}

　　public static void main(String args) throws Exception{

　　Sycn s = new Sycn ();

　　s start();

　　}

　　class Producer extends Thread{

　　public void run() {

　　try{

　　while(true){

　　int b = (int) (Math random() * );

　　System out println( Producer: a byte the value is + b);

　　pos write(b);

　　pos flush();

　　//Object o = new MyObject();

　　//oos writeObject(o);

　　//oos flush();

　　//System out println( Producer: + o);

　　}

　　}catch(Exception e){

　　//System out println(e);

　　e printStackTrace();

　　}finally{

　　try{

　　pos close();

　　pis close();

　　//oos close();

　　//ois close();

　　}catch(IOException e){

　　System out println(e);

　　}

　　}

　　}

　　}

　　class Consumer extends Thread{

　　public void run(){

　　try{

　　while(true){

　　int b = pis read();

　　System out println( Consumer: a byte the value is + String valueOf(b));

　　//Object o = ois readObject();

　　//if(o != null)

　　//System out println( Consumer: + o);

　　}

　　}catch(Exception e){

　　//System out println(e);

　　e printStackTrace();

　　}finally{

　　try{

　　pos close();

　　pis close();

　　//oos close();

　　//ois close();

　　}catch(IOException e){

　　System out println(e);

　　}

　　}

　　}

　　}

　　//class MyObject implements Serializable {

　　//}