浏览 354 次
|
精华帖 (0) :: 良好帖 (0) :: 新手帖 (0) :: 隐藏帖 (0)
|
|
|---|---|
| 作者 | 正文 |
|
最后更新时间:2008-05-13 关键字: 多线程
整个算法大致是:
多个线程同时生产短信,放到队列中(存在优先级,且先进先出), 一个消费线程从队列中拿出短信发送出去 目前出现了一个现象 生产者生产了短信,2个小时后消费者才将短信发送出去 而这2个小时的时间里面,又生产的其他短信都比较正常,都能马上发送出去,就这个停在队列里面,像是放假了不起床 并且程序运行了快半年了都没有出现这个毛病 生产者方面 public void sendSms(String xml) throws RemoteException {
log.info("Rmi接口接收到了xml"+xml);
MessageThread messageThread = Factory.getMessageThread();
messageThread.setXml(xml);
VerifyResult verifyResult = messageThread.verifyXml();
if (verifyResult.isValid()||verifyResult.equals(VerifyResult.XMLBEAN_HAS_DUPLE_PHONE)) {
log.info("过滤成功,开始放入发送队列");
new Thread(messageThread).start();
} else {
log.error("短信过滤掉了,原因:"+verifyResult.getMessage());
}
}
MessageThread run 方法
public void run() {
if (verified) {
ShortMessage shortMessage = xmlBean.convert2SM();
shortMessage.setServiceNo(shortMsgServiceNo.getServiceNo(xmlBean));
if (log4j.isDebugEnabled()) {
log4j.debug("shortMessage: 开始放入队列:" + shortMessage);
}
synchronized (queue) {
queue.push(shortMessage);
queue.notifyAll();
}
} else {
throw new ServiceException("请先检验在执行,当前没有检验");
}
}
消费者
public void run() {
stop=false;
if(log.isDebugEnabled()){
log.debug("SmsSendManager start run");
}
while (true){
ShortMessage sm =null;
synchronized (queue) {
while (queue.isEmpty()&&!stop) {
try {
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
log.error(e);
System.exit(0);
}
}
sm = queue.pop();
}
if(stop){
return;
}
if (sm == null || !sm.isFull()) {
log.info("ShortMessage is null or is not valid");
continue;
}
log.info("will sender ShortMessage:"+sm);
sendShortMessage(sm);
}
}
队列结构:
public class SmsQueue {
private LinkedList<ShortMessage>[] queuePool;
@SuppressWarnings("unchecked")
public SmsQueue(int poolSize){
if(poolSize<=0||poolSize>20)
throw new ServiceException("初始化队列出错 应该 在0-20 而实际是"+poolSize);
queuePool = new LinkedList[poolSize];
for(int i=0;i<poolSize;i++){
queuePool[i] = new LinkedList<ShortMessage>();
}
}
/**
* 把短信放到队列尾
* @param shortMessage 短信 {@link #addFirst(ShortMessage)}
*/
public void push(ShortMessage shortMessage){
if(shortMessage!=null&&shortMessage.isFull()){
int index = shortMessage.getPriority();
synchronized (queuePool) {
if (index <= 0 || index >= queuePool.length) {
queuePool[0].addLast(shortMessage);
} else {
queuePool[index].addLast(shortMessage);
}
}
}
}
/**
* 从队列头拿出短信,如果整个队列为空,返回null
* @return
*/
public ShortMessage pop(){
synchronized (queuePool) {
for (int i = queuePool.length; i > 0; i--) {
if (!queuePool[i - 1].isEmpty()) {
return queuePool[i - 1].removeFirst();
}
}
}
return null;
}
/**
* 获得队列共有多少优先级
* @return
*/
public int getQueueWidth(){
synchronized(queuePool){
return queuePool.length;
}
}
/**
* 获得当前队列状态 没有返回空 list
* @return
*/
public List<QueueStatus> getStatus(){
List<QueueStatus> retList = new ArrayList<QueueStatus>();
int i=0;
synchronized (queuePool) {
for (LinkedList<ShortMessage> queue : queuePool) {
QueueStatus status = new QueueStatus();
status.setPrority(++i);
status.setSize(queue.size());
retList.add(status);
}
}
return retList;
}
/**
* 获得所有没有发送出去的对象,发回的list 中对象引用不能修改
* 由于对象已经在内存中 所以不考虑分页
* @return
*/
public List<ShortMessage> getAllShortMessage(){
List<ShortMessage> retList = new ArrayList<ShortMessage>();
synchronized (queuePool) {
for (LinkedList<ShortMessage> queue : queuePool) {
for (ShortMessage shortMessage : queue) {
retList.add(shortMessage);
}
}
}
return Collections.unmodifiableList(retList);
}
/**
* @return id ==null 或者没有找到 返回 false
* @param id
* @return
*/
public boolean removeSMById(String id){
if(id==null)
return false;
synchronized (queuePool) {
for (LinkedList<ShortMessage> queue : queuePool) {
for (ShortMessage shortMessage : queue) {
if(shortMessage.getId().equals(id)){
queue.remove(shortMessage);
return true;
}
}
}
return false;
}
}
/**
* 对列中共有多少短信
* @return
*/
public int totalShortMessageNum(){
int size=0;
synchronized(queuePool){
for(LinkedList list:queuePool){
if(!list.isEmpty()){
size+=list.size();
}
}
return size;
}
}
/**
* 队列是否为空
* @return
*/
public boolean isEmpty() {
synchronized (queuePool) {
for (LinkedList list : queuePool) {
if (!list.isEmpty()) {
return false;
}
}
}
return true;
}
/**
* 清空队列
*
*/
public void clear(){
synchronized (queuePool) {
for (LinkedList list : queuePool) {
list.clear();
}
}
}
/**
* 把短信放到队列头 (发送短信失败的时候使用)
* @param sm
*/
public void addFirst(ShortMessage sm) {
synchronized (queuePool) {
queuePool[queuePool.length - 1].addFirst(sm);
}
}
}
声明:JavaEye文章版权属于作者,受法律保护。没有作者书面许可不得转载。
|
|
| 返回顶楼 | |


