java 信号量 semaphore 解决以下问题？

0 0

java 信号量 semaphore 解决以下问题？0

有这样一个需求，一个线程T1不停创建对象，放到数组中，数组的长度为5，当数组满了以后就将这个数组放到另一个线程线程T2，由T2将数组里的对象存到数据库中。而T1继续创建另一个数组，重复上面的操作。T2除了接收到数组，然后将数组入库以外，还有一个功能是定时5秒中做一个入库操作，5秒钟以后，不管T1创建的数组是否满5个(数组的长度为5),都要将正在创建的数组发送给T2为T2做入库操作。也就是T2接收的数组时要入库，等待5秒以后不管T1的数组是否满5个也要入库，用java 信号量 semaphore怎么做，只有T1,T2两个线程。如果不用 semaphore我也以使用生产者消费者轻松实现，但领导非要用信号量 semaphore来做，大家看怎么做？谢谢！

java多线程信号量semaphore

2012年8月24日 01:22

chenfuzhao
0
0 0 5

8个答案按时间排序按投票排序

0 0

hello，以前用观察模式做了一个引擎日志的功能，到五个执行入库时用的观察者，在添加的时候，看一下size，如果size>=5就入库。定时使用Timer，5秒钟执行以下，如果size>=5就入库。

关于同步问题，我是把添加记录和定时检查的有效代码同步的，要是嫌入库需要时间太长，我是在单起一个thread，然后复制你的数组传进去的。

至于countDownLatch，cyclicBarrier，还有你说的信号灯都没用上，同步就是synchronize的一个object

2012年10月29日 16:46

blackproof
6
0 0 7

添加评论

0 0

Java中Semaphore不能满足你的需求。
Semaphore管理一个有效的许可，类似令牌环网那个原理，如果你要控制一个资源只允许N个线程访问，这个可以使用Semaphore。

另外一个概念Latch能满足你的要求，在Java对应的实现是CountDownLatch，你可以查一下。也可以参考这个资料：
"Java concurrecncy in Practice". 5.5 Synchronizer.

2012年8月24日 14:58

lovexp2010
45
0 0 3

添加评论

0 0

引用

[img][/img][url][/url][flash=200,200][/flash]

2012年8月24日 14:28

liujian35403
30
0 0 0

添加评论

0 0

单独用信号量控制不了刚才做了简单的实验
package cn.yzc.thread;

import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.Semaphore;

/**
* @author Yao Zhichao
* @Function:
* @CreateDate :2012-8-24
* @ModifyRecord :
*/
public class SemaphoreTest {
public static final int MAX_AVAILABLE = 1;
public static final Semaphore available = new Semaphore(MAX_AVAILABLE);
public static final String[] dataStr = new String[5];

public static void main(String[] args) {
new Thread(new T1(available, dataStr)).start();
new Thread(new T2(available, dataStr)).start();
System.out.println("main thread end*");

}
}

class T1 implements Runnable {
private Semaphore available;
private String[] dataStr;
public T1(){}

public T1(Semaphore available, String[] dataStr) {
this.available = available;
this.dataStr = dataStr;
}

@Override
public void run() {
while(true) {
try {
//获取信号量
available.acquire();

for(int i=0; i<dataStr.length; i++) {
dataStr[i] = getRandString();
}

System.out.println(this.getClass().getSimpleName() + "--------" + Arrays.asList(dataStr));
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
//释放信号量
available.release();
}
}
}

public String getRandString(){
String str = "abcdefghijklmnopqrstuvwxyz1234567890";
StringBuffer sb = new StringBuffer();

for(int i=0; i<3; i++) {
sb.append(str.charAt(new Random().nextInt(str.length())));
}

return sb.toString();
}

}

class T2 implements Runnable {
private Semaphore available;
private String[] dataStr;

public T2(){}

public T2(Semaphore available, String[] dataStr) {
this.available = available;
this.dataStr = dataStr;
}

@Override
public void run() {
try {
this.available.acquire();
for(int i=0; i<dataStr.length; i++) {
System.out.println(this.getClass().getSimpleName() + "****" + Arrays.asList(dataStr));
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
this.available.release();
}
}

}
不知道写的有没有啥问题若是用condition 和 retrantlock的话那就不需要用信号量了啊

2012年8月24日 13:49

502220545
87
0 1 7

添加评论

0 0

领导非要用信号量 semaphore来做，这种事情。。。
信号量有什么好处啊，信号量这么宽松的一个同步器，里面的permits可以任意增加减少的，你用一个线程定期release，一个线程到了5个release，还得考虑多线程的问题；相对其他同步方式，没有好处，还带来问题，何必呢。

2012年8月24日 13:07

Shen.Yiyang
40
0 1 6

添加评论

0 0

提醒一下，我这里因为使用了JMS的API，不是根据时间判断的
if (batch.size() == batchNum || attemptCount == 10)
attemptCount == 10，这里是JMS侦听了10次，每次间隔1S。
你如果要处理的话，可以把这个判断做成一个Function，记录每次重置List容器的厨师时间和当前时间的一个差值，以此为判断依据。

2012年8月24日 10:29

lottons
20
0 0 0

添加评论

0 0

我前面做过这样一个需求，从一个队列不断地接收消息，当消息数量达到一定的数量的时候进行批量入库。当消息没有达到这个数量的时候，如果间隔时间达到一定的时间也会进行入库操作。接收消息的线程会不断地创建一个List容器，存放接收到的消息。当List达到一个阀值的时候，提交给另一个线程入库。当List没有达到这个阀值，但是连续10s没有获取到新消息的时候，同样提交给另一个线程做批量操作，将消息缓存中缓存的消息进行处理，防止这部分消息因为没有后续消息的触发而堆积在缓存中。
实例代码，接受消息创建List的部分：
while (true) {
Object message = receiveMsg();

if (message != null){
log.debug("receive message : " + message);
batch.add(message);
}else {
attemptCount ++;
}

// 批量接受一批消息之后进行处理
// 进行批量处理的条件：1、批量处理缓存消息中的数量达到要求的数量；2、一定时间段内没有收到消息时直接触发
if (batch.size() == batchNum || attemptCount == 10) {
log.info(" ++ process batch message: " + batch.size());
// 将批量消息缓存区中的消息转换成字符串消息数组，用于后续处理
String[] tms = new String[batch.size()];
for (int i = 0; i < batch.size(); i++) {
tms[i] = transformMsg(batch.get(i));
log.debug("add message : " + tms[i]);
}
// 待处理的消息缓存中有消息的话，就通过线程池启动运行处理flow
if (tms.length > 0){
ProcessExecuteTask pet = new ProcessExecuteTask();
pet.setMessage(tms);

pet.setSequenceProcessor((SequenceProcessor) processor);

executor.doExecute(pet);
batch.clear();
}
// 处理的标志lock文件存在，表示继续处理。否则，表示需要停止运行处理。
if (!flock.exists())
break;
attemptCount = 0;
}
}

2012年8月24日 10:26

lottons
20
0 0 0

添加评论

0 0

需求：
1、数组大小5
2、T2线程5秒自动同步一次（不管数组是否满）
3、数组满时，不管到不到5秒都需要主动触发T2保存

可以使用信号量：
但需要三个线程：
1、T2在等待信号量
2、T1 生成数组数据当生产到5时释放信号量
3、T3每隔5秒释放信号量
4、T4获取信号量之后执行完任务后再等待信号量

其实可以使用wait notify
1、T2在wait
2、T1 生成数组数据当生产到5时 notify
3、T3每隔5秒 notify
4、T4获取信号量之后执行完任务后再 wait