(1)本实验的多个缓冲区不是环形循环的，也不要求按顺序访问。生产者可以把产品放到目前某一个空缓冲区中。
(2)消费者只消费指定生产者的产品。
(3)在测试用例文件中指定了所有的生产和消费的需求，只有当共享缓冲区的数据满足了所有关于它的消费需求后，此共享缓冲区才可以作为空闲空间允许新的生产者使用。
(4)本实验在为生产者分配缓冲区时各生产者间必须互斥，此后各个生产者的具体生产活动可以并发。而消费者之间只有在对同一产品进行消费时才需要互斥，同时它们在消费过程结束时需要判断该消费对象是否已经消费完毕并清除该产品。
Windows
用来实现同步和互斥的实体。在Windows
中，常见的同步对象有：信号量(Semaphore)、
互斥量(Mutex)、临界段(CriticalSection)和事件(Event)等。本程序中用到了前三个。使用这些对象都分
为三个步骤，一是创建或者初始化：接着请求该同步对象，随即进入临界区，这一步对应于互斥量的
上锁；最后释放该同步对象，这对应于互斥量的解锁。这些同步对象在一个线程中创建，在其他线程
中都可以使用，从而实现同步互斥。当然，在进程间使用这些同步对象实现同步的方法是类似的。
1．用锁操作原语实现互斥
为解决进程互斥进人临界区的问题，可为每类临界区设置一把锁，该锁有打开和关闭两种状态，进程执行临界区程序的操作按下列步骤进行：
①关锁。先检查锁的状态，如为关闭状态，则等待其打开；如已打开了，则将其关闭，继续执行步骤②的操作。
②执行临界区程序。
③开锁。将锁打开，退出临界区。
2．信号量及WAIT，SIGNAL操作原语
信号量的初值可以由系统根据资源情况和使用需要来确定。在初始条件下信号量的指针项可以置为0，表示队列为空。信号量在使用过程中它的值是可变的，但只能由WAIT，SIGNAL操作来改变。设信号量为S，对S的WAIT操作记为WAIT（S），对它的SIGNAL操作记为SIGNAL（S）。
WAIT（S）：顺序执行以下两个动作：
①信号量的值减1，即S=S－1；
②如果S≥0，则该进程继续执行；
如果
S（0，则把该进程的状态置为阻塞态，把相应的WAITCB连人该信号量队列的末尾，并放弃处理机，进行等待（直至其它进程在S上执行SIGNAL操作，把它释放出来为止）。
SIGNAL（S）：顺序执行以下两个动作
①S值加
1，即
S=S＋1；
②如果S）0，则该进程继续运行；
如果S（0则释放信号量队列上的第一个PCB（既信号量指针项所指向的PCB）所对应的进程（把阻塞态改为就绪态），执行SIGNAL操作的进程继续运行。
在具体实现时注意，WAIT，SIGNAL操作都应作为一个整体实施，不允许分割或相互穿插执行。也就是说，WAIT，SIGNAL操作各自都好像对应一条指令，需要不间断地做下去，否则会造成混乱。
从物理概念上讲，信号量S）时，S值表示可用资源的数量。执行一次WAIT操作意味着请求分配一个单位资源，因此S值减1；当S<0时，表示已无可用资源，请求者必须等待别的进程释放了该类资源，它才能运行下去。所以它要排队。而执行一次SIGNAL操作意味着释放一个单位资源，因此S值加1；若S（0时，表示有某些进程正在等待该资源，因而要把队列头上的进程唤醒，释放资源的进程总是可以运行下去的。
---------------
/**
*
生产者
*
*/
public
class
Producer
implements
Runnable{
private
Semaphore
mutex,full,empty
private
Buffer
buf
String
name
public
Producer(String
name,Semaphore
mutex,Semaphore
full,Semaphore
empty,Buffer
buf){
this.mutex
=
mutex
this.full
=
full
this.empty
=
empty
this.buf
=
buf
this.name
=
name
}
public
void
run(){
while(true){
empty.p()
mutex.p()
System.out.println(name+"
inserts
a
new
product
into
"+buf.nextEmptyIndex)
buf.nextEmptyIndex
=
(buf.nextEmptyIndex+1)%buf.size
mutex.v()
full.v()
try
{
Thread.sleep(1000)
}
catch
(InterruptedException
e)
{
e.printStackTrace()
}
}
}
}
---------------
/**
*
消费者
*
*/
public
class
Customer
implements
Runnable{
private
Semaphore
mutex,full,empty
private
Buffer
buf
String
name
public
Customer(String
name,Semaphore
mutex,Semaphore
full,Semaphore
empty,Buffer
buf){
this.mutex
=
mutex
this.full
=
full
this.empty
=
empty
this.buf
=
buf
this.name
=
name
}
public
void
run(){
while(true){
full.p()
mutex.p()
System.out.println(name+"
gets
a
product
from
"+buf.nextFullIndex)
buf.nextFullIndex
=
(buf.nextFullIndex+1)%buf.size
mutex.v()
empty.v()
try
{
Thread.sleep(1000)
}
catch
(InterruptedException
e)
{
e.printStackTrace()
}
}
}
}
-------------------------
/**
*
缓冲区
*
*/
public
class
Buffer{
public
Buffer(int
size,int
nextEmpty,int
nextFull){
this.nextEmptyIndex
=
nextEmpty
this.nextFullIndex
=
nextFull
this.size
=
size
}
public
int
size
public
int
nextEmptyIndex
public
int
nextFullIndex
}
-----------------
/**
*
此类用来模拟信号量
*
*/
public
class
Semaphore{
private
int
semValue
public
Semaphore(int
semValue){
this.semValue
=
semValue
}
public
synchronized
void
p(){
semValue--
if(semValue<0){
try
{
this.wait()
}
catch
(InterruptedException
e)
{
e.printStackTrace()
}
}
}
public
synchronized
void
v(){
semValue++
if(semValue<=0){
this.notify()
}
}
}
------------------------
public
class
Test
extends
Thread
{
public
static
void
main(String[]
args)
{
Buffer
bf=new
Buffer(10,0,0)
Semaphore
mutex=new
Semaphore(1)
Semaphore
full=new
Semaphore(0)
Semaphore
empty=new
Semaphore(10)
//new
Thread(new
Producer("p001",mutex,full,empty,bf)).start()
Producer
p=new
Producer("p001",mutex,full,empty,bf)
new
Thread(new
Producer("p002",mutex,full,empty,bf)).start()
new
Thread(new
Producer("p003",mutex,full,empty,bf)).start()
new
Thread(new
Producer("p004",mutex,full,empty,bf)).start()
new
Thread(new
Producer("p005",mutex,full,empty,bf)).start()
try{
sleep(3000)
}
catch(Exception
ex)
{
ex.printStackTrace()
}
new
Thread(new
Customer("c001",mutex,full,empty,bf)).start()
new
Thread(new
Customer("c002",mutex,full,empty,bf)).start()
new
Thread(new
Customer("c003",mutex,full,empty,bf)).start()
new
Thread(new
Customer("c004",mutex,full,empty,bf)).start()
new
Thread(new
Customer("c005",mutex,full,empty,bf)).start()
}
}
--------------------------------------------
linux中的进程通信分为三个部分：低级通信，管道通信和进程间通信IPC（inter process communication）。linux的低级通信主要用来传递进程的控制信号——文件锁和软中断信号机制。linux的进程间通信IPC有三个部分——①信号量，②共享内存和③消息队列。以下是我编写的linux进程通信的C语言实现代码。操作系统为redhat9.0，编辑器为vi，编译器采用gcc。下面所有实现代码均已经通过测试，运行无误。一.低级通信--信号通信
signal.c
#include
#include
#include
/*捕捉到信号sig之后，执行预先预定的动作函数*/
void sig_alarm(int sig)
{
printf("---the signal received is %d. /n", sig)
signal(SIGINT, SIG_DFL)//SIGINT终端中断信号，SIG_DFL：恢复默认行为，SIN_IGN：忽略信号
}
int main()
{
signal(SIGINT, sig_alarm)//捕捉终端中断信号
while(1)
{
printf("waiting here!/n")
sleep(1)
}
return 0
}
二.管道通信
pipe.c
#include
#define BUFFER_SIZE 30
int main()
{
int x
int fd[2]
char buf[BUFFER_SIZE]
char s[BUFFER_SIZE]
pipe(fd)//创建管道
while((x=fork())==-1)//创建管道失败时，进入循环
/*进入子进程，子进程向管道中写入一个字符串*/
if(x==0)
{
sprintf(buf,"This is an example of pipe!/n")
write(fd[1],buf,BUFFER_SIZE)
exit(0)
}
/*进入父进程，父进程从管道的另一端读出刚才写入的字符串*/
else
{
wait(0)//等待子进程结束
read(fd[0],s,BUFFER_SIZE)//读出字符串，并将其储存在char s[]中
printf("%s",s)//打印字符串
}
return 0
}
三.进程间通信——IPC
①信号量通信
sem.c
#include
#include
#include
#include types.h>
#include ipc.h>
#include sem.h>
/*联合体变量*/
union semun
{
int val//信号量初始值
struct semid_ds *buf
unsigned short int *array
struct seminfo *__buf
}
/*函数声明，信号量定义*/
static int set_semvalue(void)//设置信号量
static void del_semvalue(void)//删除信号量
static int semaphore_p(void)//执行P操作
static int semaphore_v(void)//执行V操作
static int sem_id//信号量标识符
int main(int argc, char *argv[])
{
int i
int pause_time
char op_char = 'O'
srand((unsigned int)getpid())
sem_id = semget((key_t)1234, 1, 0666 | IPC_CREAT)//创建一个信号量,IPC_CREAT表示创建一个新的信号量
/*如果有参数，设置信号量，修改字符*/
if (argc >1)
{
if (!set_semvalue())
{
fprintf(stderr, "Failed to initialize semaphore/n")
exit(EXIT_FAILURE)
}
op_char = 'X'
sleep(5)
}
for(i = 0i <10i++)
{
/*执行P操作*/
if (!semaphore_p())
exit(EXIT_FAILURE)
printf("%c", op_char)
fflush(stdout)
pause_time = rand() % 3
sleep(pause_time)
printf("%c", op_char)
fflush(stdout)
/*执行V操作*/
if (!semaphore_v())
exit(EXIT_FAILURE)
pause_time = rand() % 2
sleep(pause_time)
}
printf("/n%d - finished/n", getpid())
if (argc >1)
{
sleep(10)
del_semvalue()//删除信号量
}
exit(EXIT_SUCCESS)
}
/*设置信号量*/
static int set_semvalue(void)
{
union semun sem_union
sem_union.val = 1
if (semctl(sem_id, 0, SETVAL, sem_union) == -1)
return(0)
return(1)
}
/*删除信号量*/
static void del_semvalue(void)
{
union semun sem_union
if (semctl(sem_id, 0, IPC_RMID, sem_union) == -1)
fprintf(stderr, "Failed to delete semaphore/n")
}
/*执行P操作*/
static int semaphore_p(void)
{
struct sembuf sem_b
sem_b.sem_num = 0
sem_b.sem_op = -1/* P() */
sem_b.sem_flg = SEM_UNDO
if (semop(sem_id, &sem_b, 1) == -1)
{
fprintf(stderr, "semaphore_p failed/n")
return(0)
}
return(1)
}
/*执行V操作*/
static int semaphore_v(void)
{
struct sembuf sem_b
sem_b.sem_num = 0
sem_b.sem_op = 1/* V() */
sem_b.sem_flg = SEM_UNDO
if (semop(sem_id, &sem_b, 1) == -1)
{
fprintf(stderr, "semaphore_v failed/n")
return(0)
}
return(1)
}
②消息队列通信
send.c
#include
#include
#include
#include
#include
#include types.h>
#include ipc.h>
#include msg.h>
#define MAX_TEXT 512
/*用于消息收发的结构体--my_msg_type：消息类型，some_text：消息正文*/
struct my_msg_st
{
long int my_msg_type
char some_text[MAX_TEXT]
}
int main()
{
int running = 1//程序运行标识符
struct my_msg_st some_data
int msgid//消息队列标识符
char buffer[BUFSIZ]
/*创建与接受者相同的消息队列*/
msgid = msgget((key_t)1234, 0666 | IPC_CREAT)
if (msgid == -1)
{
fprintf(stderr, "msgget failed with error: %d/n", errno)
exit(EXIT_FAILURE)
}
/*向消息队列中发送消息*/
while(running)
{
printf("Enter some text: ")
fgets(buffer, BUFSIZ, stdin)
some_data.my_msg_type = 1
strcpy(some_data.some_text, buffer)
if (msgsnd(msgid, (void *)&some_data, MAX_TEXT, 0) == -1)
{
fprintf(stderr, "msgsnd failed/n")
exit(EXIT_FAILURE)
}
if (strncmp(buffer, "end", 3) == 0)
{
running = 0
}
}
exit(EXIT_SUCCESS)
}
receive.c
#include
#include
#include
#include
#include
#include types.h>
#include ipc.h>
#include msg.h>
/*用于消息收发的结构体--my_msg_type：消息类型，some_text：消息正文*/
struct my_msg_st
{
long int my_msg_type
char some_text[BUFSIZ]
}
int main()
{
int running = 1//程序运行标识符
int msgid//消息队列标识符
struct my_msg_st some_data
long int msg_to_receive = 0//接收消息的类型--0表示msgid队列上的第一个消息
/*创建消息队列*/
msgid = msgget((key_t)1234, 0666 | IPC_CREAT)
if (msgid == -1)
{
fprintf(stderr, "msgget failed with error: %d/n", errno)
exit(EXIT_FAILURE)
}
/*接收消息*/
while(running)
{
if (msgrcv(msgid, (void *)&some_data, BUFSIZ,msg_to_receive, 0) == -1)
{
fprintf(stderr, "msgrcv failed with error: %d/n", errno)
exit(EXIT_FAILURE)
}
printf("You wrote: %s", some_data.some_text)
if (strncmp(some_data.some_text, "end", 3) == 0)
{
running = 0
}
}
/*删除消息队列*/
if (msgctl(msgid, IPC_RMID, 0) == -1)
{
fprintf(stderr, "msgctl(IPC_RMID) failed/n")
exit(EXIT_FAILURE)
}
exit(EXIT_SUCCESS)
}
③共享内存通信
share.h
#define TEXT_SZ 2048 //申请共享内存大小
struct shared_use_st
{
int written_by_you//written_by_you为1时表示有数据写入，为0时表示数据已经被消费者提走
char some_text[TEXT_SZ]
}
producer.c
#include
#include
#include
#include
#include types.h>
#include ipc.h>
#include shm.h>
#include "share.h"
int main()
{
int running = 1//程序运行标志位
void *shared_memory = (void *)0
struct shared_use_st *shared_stuff
char buffer[BUFSIZ]
int shmid//共享内存标识符
/*创建共享内存*/
shmid = shmget((key_t)1234, sizeof(struct shared_use_st), 0666 | IPC_CREAT)
if (shmid == -1)
{
fprintf(stderr, "shmget failed/n")
exit(EXIT_FAILURE)
}
/*将共享内存连接到一个进程的地址空间中*/
shared_memory = shmat(shmid, (void *)0, 0)//指向共享内存第一个字节的指针
if (shared_memory == (void *)-1)
{
fprintf(stderr, "shmat failed/n")
exit(EXIT_FAILURE)
}
printf("Memory attached at %X/n", (int)shared_memory)
shared_stuff = (struct shared_use_st *)shared_memory
/*生产者写入数据*/
while(running)
{
while(shared_stuff->written_by_you == 1)
{
sleep(1)
printf("waiting for client.../n")
}
printf("Enter some text: ")
fgets(buffer, BUFSIZ, stdin)
strncpy(shared_stuff->some_text, buffer, TEXT_SZ)
shared_stuff->written_by_you = 1
if (strncmp(buffer, "end", 3) == 0)
{
running = 0
}
}
/*该函数用来将共享内存从当前进程中分离,仅使得当前进程不再能使用该共享内存*/
if (shmdt(shared_memory) == -1)
{
fprintf(stderr, "shmdt failed/n")
exit(EXIT_FAILURE)
}
printf("producer exit./n")
exit(EXIT_SUCCESS)
}
customer.c
#include
#include
#include
#include
#include types.h>
#include ipc.h>
#include shm.h>
#include "share.h"
int main()
{
int running = 1//程序运行标志位
void *shared_memory = (void *)0
struct shared_use_st *shared_stuff
int shmid//共享内存标识符
srand((unsigned int)getpid())
/*创建共享内存*/
shmid = shmget((key_t)1234, sizeof(struct shared_use_st), 0666 | IPC_CREAT)
if (shmid == -1)
{
fprintf(stderr, "shmget failed/n")
exit(EXIT_FAILURE)
}
/*将共享内存连接到一个进程的地址空间中*/
shared_memory = shmat(shmid, (void *)0, 0)//指向共享内存第一个字节的指针
if (shared_memory == (void *)-1)
{
fprintf(stderr, "shmat failed/n")
exit(EXIT_FAILURE)
}
printf("Memory attached at %X/n", (int)shared_memory)
shared_stuff = (struct shared_use_st *)shared_memory
shared_stuff->written_by_you = 0
/*消费者读取数据*/
while(running)
{
if (shared_stuff->written_by_you)
{
printf("You wrote: %s", shared_stuff->some_text)
sleep( rand() % 4 )
shared_stuff->written_by_you = 0
if (strncmp(shared_stuff->some_text, "end", 3) == 0)
{
running = 0
}
}
}
/*该函数用来将共享内存从当前进程中分离,仅使得当前进程不再能使用该共享内存*/
if (shmdt(shared_memory) == -1)
{
fprintf(stderr, "shmdt failed/n")
exit(EXIT_FAILURE)
}
/*将共享内存删除,所有进程均不能再访问该共享内存*/
if (shmctl(shmid, IPC_RMID, 0) == -1)
{
fprintf(stderr, "shmctl(IPC_RMID) failed/n")
exit(EXIT_FAILURE)
}
exit(EXIT_SUCCESS)
}
摘自：
欢迎分享，转载请注明来源：夏雨云
微信扫一扫
支付宝扫一扫
评论列表（0条）